Handle reentrant case in Room's acquireTransactionThread() During a suspending transaction Room uses a thread from the transaction executor, starts an event loop in it and dispatches suspending database operations to it so they are all encapsulated within the transaction coroutine. It is usually expected that the transaction thread is different from the one starting the transaction, but in some cases they are the same. To handle such reentrant cases the `withTransaction()` has been refactored to no longer rely joining on a control job and instead it will execute the suspending transaction block from within the `runBlocking` in the transaction thread. Executing the suspending transaction block in the `runBlocking` prevents a deadlock situation due to the continuation bouncing between the caller to the transaction thread and back. The deadlock specifically happened when Room started the event loop via `runBlocking`, resumed the `acquireTransactionThread()` coroutine but also joined the `runBlocking` coroutine with the control job. These set of events would cause a deadlock when both the acquire thread and transaction thread are the same because `acquireTransactionThread()` is unable to continue as it is waiting for `runBlocking` to finish which in turns waits for the control job to be done. Meanwhile the job never completes because the database transaction does not finish, i.e. `TransactionElement.release` is never called. By eliminating the need to wait on a control job but still support nested transactions via the transaction context, Room is able to avoid the deadlock situation. This behaviour was mostly observed in tests where the transaction executor is usually an instant executor, but it could happen in an actual application if Room is configured to share a thread pool executor as the transaction executor and a background coroutine dispatcher and the thread pool is sufficiently busy that only one thread is left for both starting the transaction (to acquire the transaction thread) and the transaction thread itself. Bug: 251503005 Test: SuspendingQueryTest Change-Id: I319d977a080b89880d5cf68d139c69a8ba096107 (cherry picked from commit on android-review.googlesource.com host: 7d8a5b162ba1a8814da51ba950b8555ca94895b3) Merged-In: I319d977a080b89880d5cf68d139c69a8ba096107 (cherry picked from commit on android-review.googlesource.com host: a613c4f3691d3cb9ae3b9f869778178ab76be230) Merged-In: I319d977a080b89880d5cf68d139c69a8ba096107
diff --git a/room/integration-tests/kotlintestapp/src/androidTest/java/androidx/room/integration/kotlintestapp/TestDatabase.kt b/room/integration-tests/kotlintestapp/src/androidTest/java/androidx/room/integration/kotlintestapp/TestDatabase.kt index b3cb2b9..589d690 100644 --- a/room/integration-tests/kotlintestapp/src/androidTest/java/androidx/room/integration/kotlintestapp/TestDatabase.kt +++ b/room/integration-tests/kotlintestapp/src/androidTest/java/androidx/room/integration/kotlintestapp/TestDatabase.kt
@@ -18,7 +18,9 @@ import androidx.room.Database import androidx.room.RoomDatabase +import androidx.room.androidx.room.integration.kotlintestapp.dao.CounterDao import androidx.room.androidx.room.integration.kotlintestapp.dao.UsersDao +import androidx.room.androidx.room.integration.kotlintestapp.vo.Counter import androidx.room.androidx.room.integration.kotlintestapp.vo.User import androidx.room.integration.kotlintestapp.dao.AbstractDao import androidx.room.integration.kotlintestapp.dao.BooksDao @@ -37,7 +39,7 @@ entities = [ Book::class, Author::class, Publisher::class, BookAuthor::class, NoArgClass::class, DataClassFromDependency::class, JavaEntity::class, - EntityWithJavaPojoList::class, User::class + EntityWithJavaPojoList::class, User::class, Counter::class ], version = 1, exportSchema = false @@ -53,4 +55,6 @@ abstract fun dependencyDao(): DependencyDao abstract fun abstractDao(): AbstractDao + + abstract fun counterDao(): CounterDao }
diff --git a/room/integration-tests/kotlintestapp/src/androidTest/java/androidx/room/integration/kotlintestapp/dao/BooksDao.kt b/room/integration-tests/kotlintestapp/src/androidTest/java/androidx/room/integration/kotlintestapp/dao/BooksDao.kt index d53132c..7ec25bd 100644 --- a/room/integration-tests/kotlintestapp/src/androidTest/java/androidx/room/integration/kotlintestapp/dao/BooksDao.kt +++ b/room/integration-tests/kotlintestapp/src/androidTest/java/androidx/room/integration/kotlintestapp/dao/BooksDao.kt
@@ -136,6 +136,9 @@ @Query("SELECT * FROM book") suspend fun getBooksSuspend(): List<Book> + @Query("SELECT * FROM publisher") + suspend fun getPublishersSuspend(): List<Publisher> + @Query("UPDATE book SET salesCnt = salesCnt + 1 WHERE bookId = :bookId") fun increaseBookSales(bookId: String)
diff --git a/room/integration-tests/kotlintestapp/src/androidTest/java/androidx/room/integration/kotlintestapp/dao/CounterDao.kt b/room/integration-tests/kotlintestapp/src/androidTest/java/androidx/room/integration/kotlintestapp/dao/CounterDao.kt new file mode 100644 index 0000000..62e7709 --- /dev/null +++ b/room/integration-tests/kotlintestapp/src/androidTest/java/androidx/room/integration/kotlintestapp/dao/CounterDao.kt
@@ -0,0 +1,31 @@ +/* + * Copyright 2022 The Android Open Source Project + * + * Licensed 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 androidx.room.androidx.room.integration.kotlintestapp.dao + +import androidx.room.Dao +import androidx.room.Query +import androidx.room.Upsert +import androidx.room.androidx.room.integration.kotlintestapp.vo.Counter + +@Dao +interface CounterDao { + @Upsert + suspend fun upsert(c: Counter) + + @Query("SELECT * FROM Counter WHERE id = :id") + suspend fun getCounter(id: Long): Counter +} \ No newline at end of file
diff --git a/room/integration-tests/kotlintestapp/src/androidTest/java/androidx/room/integration/kotlintestapp/test/SuspendingQueryTest.kt b/room/integration-tests/kotlintestapp/src/androidTest/java/androidx/room/integration/kotlintestapp/test/SuspendingQueryTest.kt index 48efc91..29a48db 100644 --- a/room/integration-tests/kotlintestapp/src/androidTest/java/androidx/room/integration/kotlintestapp/test/SuspendingQueryTest.kt +++ b/room/integration-tests/kotlintestapp/src/androidTest/java/androidx/room/integration/kotlintestapp/test/SuspendingQueryTest.kt
@@ -16,10 +16,13 @@ package androidx.room.integration.kotlintestapp.test +import android.content.Context import android.os.Build import androidx.arch.core.executor.ArchTaskExecutor +import androidx.arch.core.executor.TaskExecutor import androidx.room.Room import androidx.room.RoomDatabase +import androidx.room.androidx.room.integration.kotlintestapp.vo.Counter import androidx.room.integration.kotlintestapp.NewThreadDispatcher import androidx.room.integration.kotlintestapp.TestDatabase import androidx.room.integration.kotlintestapp.vo.Book @@ -32,8 +35,22 @@ import androidx.test.filters.LargeTest import com.google.common.truth.Truth.assertThat import com.google.common.truth.Truth.assertWithMessage +import java.io.IOException +import java.util.concurrent.CountDownLatch +import java.util.concurrent.ExecutorService +import java.util.concurrent.Executors +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger +import kotlin.coroutines.intrinsics.COROUTINE_SUSPENDED +import kotlin.coroutines.intrinsics.intercepted +import kotlin.coroutines.intrinsics.suspendCoroutineUninterceptedOrReturn +import kotlin.coroutines.resume +import kotlinx.coroutines.DelicateCoroutinesApi import kotlinx.coroutines.Dispatchers -import kotlinx.coroutines.ObsoleteCoroutinesApi +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.Runnable +import kotlinx.coroutines.TimeoutCancellationException +import kotlinx.coroutines.asCoroutineDispatcher import kotlinx.coroutines.async import kotlinx.coroutines.cancelAndJoin import kotlinx.coroutines.coroutineScope @@ -41,18 +58,13 @@ import kotlinx.coroutines.launch import kotlinx.coroutines.newSingleThreadContext import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.test.runTest import kotlinx.coroutines.withContext +import kotlinx.coroutines.withTimeout import org.junit.After import org.junit.Assert.fail import org.junit.Test import org.junit.runner.RunWith -import java.io.IOException -import java.util.concurrent.CountDownLatch -import java.util.concurrent.ExecutorService -import java.util.concurrent.Executors -import java.util.concurrent.TimeUnit -import java.util.concurrent.atomic.AtomicInteger -import kotlinx.coroutines.DelicateCoroutinesApi @LargeTest @RunWith(AndroidJUnit4::class) @@ -158,9 +170,7 @@ database.endTransaction() } } - runBlocking { - assertThat(booksDao.getBooksSuspend()).isEqualTo(listOf(TestUtil.BOOK_2)) - } + assertThat(booksDao.allBooks).isEqualTo(listOf(TestUtil.BOOK_2)) } @Test @@ -182,9 +192,7 @@ database.endTransaction() } } - runBlocking { - assertThat(booksDao.getBooksSuspend()).isEqualTo(listOf(TestUtil.BOOK_2)) - } + assertThat(booksDao.allBooks).isEqualTo(listOf(TestUtil.BOOK_2)) } @Test @@ -206,9 +214,7 @@ database.endTransaction() } } - runBlocking(NewThreadDispatcher()) { - assertThat(booksDao.getBooksSuspend()).isEqualTo(listOf(TestUtil.BOOK_2)) - } + assertThat(booksDao.allBooks).isEqualTo(listOf(TestUtil.BOOK_2)) } @Test @@ -256,10 +262,25 @@ booksDao.deleteUnsoldBooks() } } - runBlocking(NewThreadDispatcher()) { - assertThat(booksDao.getBooksSuspend()) - .isEqualTo(listOf(TestUtil.BOOK_2)) + assertThat(booksDao.allBooks).isEqualTo(listOf(TestUtil.BOOK_2)) + } + + @Test + fun withTransaction_withContext_newThreadDispatcher() { + runBlocking { + withContext(NewThreadDispatcher()) { + database.withTransaction { + booksDao.insertPublisherSuspend( + TestUtil.PUBLISHER.publisherId, + TestUtil.PUBLISHER.name + ) + booksDao.insertBookSuspend(TestUtil.BOOK_1.copy(salesCnt = 0)) + booksDao.insertBookSuspend(TestUtil.BOOK_2) + booksDao.deleteUnsoldBooks() + } + } } + assertThat(booksDao.allBooks).isEqualTo(listOf(TestUtil.BOOK_2)) } @Test @@ -275,10 +296,7 @@ booksDao.deleteUnsoldBooks() } } - runBlocking(NewThreadDispatcher()) { - assertThat(booksDao.getBooksSuspend()) - .isEqualTo(listOf(TestUtil.BOOK_2)) - } + assertThat(booksDao.allBooks).isEqualTo(listOf(TestUtil.BOOK_2)) } @Test @@ -301,6 +319,31 @@ } @Test + fun withTransaction_contextSwitch_exception() { + runBlocking { + try { + database.withTransaction { + booksDao.insertPublisherSuspend( + TestUtil.PUBLISHER.publisherId, + TestUtil.PUBLISHER.name + ) + withContext(Dispatchers.IO) { + booksDao.insertBookSuspend(TestUtil.BOOK_1.copy(salesCnt = 0)) + booksDao.insertBookSuspend(TestUtil.BOOK_2) + } + booksDao.deleteUnsoldBooks() + throw IOException("Boom!") + } + } catch (ex: IOException) { + assertThat(ex).hasMessageThat() + .contains("Boom") + } + assertThat(booksDao.getPublishersSuspend()).isEmpty() + assertThat(booksDao.getBooksSuspend()).isEmpty() + } + } + + @Test fun withTransaction_exception() { runBlocking { database.withTransaction { @@ -314,6 +357,7 @@ try { database.withTransaction { booksDao.insertBookSuspend(TestUtil.BOOK_2) + booksDao.insertBookSuspend(TestUtil.BOOK_3) throw IOException("Boom!") } @Suppress("UNREACHABLE_CODE") @@ -369,8 +413,8 @@ } } - assertThat(booksDao.getBooksSuspend()) - .isEqualTo(emptyList<Book>()) + assertThat(booksDao.getPublishersSuspend()).isEmpty() + assertThat(booksDao.getBooksSuspend()).isEmpty() } } @@ -464,6 +508,7 @@ @Test fun withTransaction_cancelCoroutine() { + runBlocking { booksDao.insertPublisherSuspend( TestUtil.PUBLISHER.publisherId, @@ -496,6 +541,42 @@ } @Test + fun withTransaction_busyExecutor_cancelCoroutine() { + val executorService = Executors.newSingleThreadExecutor() + val localDatabase = Room.inMemoryDatabaseBuilder( + ApplicationProvider.getApplicationContext(), TestDatabase::class.java + ) + .setTransactionExecutor(executorService) + .build() + + // Simulate a busy executor, no thread to acquire for transaction. + val busyLatch = CountDownLatch(1) + executorService.execute { + busyLatch.await() + } + runBlocking { + val startedRunning = CountDownLatch(1) + val job = launch(Dispatchers.IO) { + startedRunning.countDown() + delay(200) // yield and delay to queue the runnable in transaction executor + localDatabase.withTransaction { + fail("Transaction block should have never run!") + } + } + + assertThat(startedRunning.await(1, TimeUnit.SECONDS)).isTrue() + job.cancelAndJoin() + } + + // free busy thread + busyLatch.countDown() + executorService.shutdown() + assertThat(executorService.awaitTermination(1, TimeUnit.SECONDS)).isTrue() + + assertThat(localDatabase.booksDao().getPublishers()).isEmpty() + } + + @Test fun withTransaction_blockingDaoMethods() { runBlocking { database.withTransaction { @@ -655,15 +736,12 @@ } } - runBlocking { - // as Set since insertion order is undefined - assertThat(booksDao.getBooksSuspend().toSet()) - .isEqualTo(setOf(TestUtil.BOOK_1, TestUtil.BOOK_2)) - } + // as Set since insertion order is undefined + assertThat(booksDao.allBooks.toSet()) + .isEqualTo(setOf(TestUtil.BOOK_1, TestUtil.BOOK_2)) } @Test - @ObsoleteCoroutinesApi @Suppress("DeferredResultUnused") fun withTransaction_multipleTransactions_multipleThreads() { runBlocking { @@ -689,31 +767,16 @@ } } - runBlocking { - // as Set since insertion order is undefined - assertThat(booksDao.getBooksSuspend().toSet()) - .isEqualTo(setOf(TestUtil.BOOK_1, TestUtil.BOOK_2)) - } + // as Set since insertion order is undefined + assertThat(booksDao.allBooks.toSet()) + .isEqualTo(setOf(TestUtil.BOOK_1, TestUtil.BOOK_2)) } @Test @Suppress("DeferredResultUnused") fun withTransaction_multipleTransactions_verifyThreadUsage() { val busyThreadsCount = AtomicInteger() - // Executor wrapper that counts threads that are busy executing commands. - class WrappedService(val delegate: ExecutorService) : ExecutorService by delegate { - override fun execute(command: Runnable) { - delegate.execute { - busyThreadsCount.incrementAndGet() - try { - command.run() - } finally { - busyThreadsCount.decrementAndGet() - } - } - } - } - val wrappedExecutor = WrappedService(Executors.newCachedThreadPool()) + val wrappedExecutor = BusyCountingService(busyThreadsCount, Executors.newCachedThreadPool()) val localDatabase = Room.inMemoryDatabaseBuilder( ApplicationProvider.getApplicationContext(), TestDatabase::class.java ) @@ -740,57 +803,67 @@ } } - wrappedExecutor.awaitTermination(1, TimeUnit.SECONDS) + assertThat(busyThreadsCount.get()).isEqualTo(0) + wrappedExecutor.shutdown() + assertThat(wrappedExecutor.awaitTermination(1, TimeUnit.SECONDS)).isTrue() } @Test fun withTransaction_busyExecutor() { + val executorService = Executors.newSingleThreadExecutor() + val localDatabase = Room.inMemoryDatabaseBuilder( + ApplicationProvider.getApplicationContext(), TestDatabase::class.java + ) + .setTransactionExecutor(executorService) + .build() + + // Simulate a busy executor, no thread to acquire for transaction. + val busyLatch = CountDownLatch(1) + executorService.execute { + busyLatch.await() + } runBlocking { - val executorService = Executors.newSingleThreadExecutor() - val localDatabase = Room.inMemoryDatabaseBuilder( - ApplicationProvider.getApplicationContext(), TestDatabase::class.java - ) - .setTransactionExecutor(executorService) - .build() - - // Simulate a busy executor, no thread to acquire for transaction. - val busyLatch = CountDownLatch(1) - executorService.execute { - busyLatch.await() - } - var asyncExecuted = false - val transactionLatch = CountDownLatch(1) val job = async(Dispatchers.IO) { asyncExecuted = true localDatabase.withTransaction { - transactionLatch.countDown() + booksDao.insertPublisherSuspend( + TestUtil.PUBLISHER.publisherId, + TestUtil.PUBLISHER.name + ) } } - assertThat(transactionLatch.await(1000, TimeUnit.MILLISECONDS)).isFalse() + try { + withTimeout(1000) { + job.join() + } + fail("A timeout should have occurred!") + } catch (_: TimeoutCancellationException) { } job.cancelAndJoin() assertThat(asyncExecuted).isTrue() - - // free busy thread - busyLatch.countDown() - executorService.awaitTermination(1, TimeUnit.SECONDS) } + // free busy thread + busyLatch.countDown() + executorService.shutdown() + assertThat(executorService.awaitTermination(1, TimeUnit.SECONDS)).isTrue() + + assertThat(booksDao.getPublishers()).isEmpty() } @Test fun withTransaction_shutdownExecutor() { + val executorService = Executors.newCachedThreadPool() + val localDatabase = Room.inMemoryDatabaseBuilder( + ApplicationProvider.getApplicationContext(), TestDatabase::class.java + ) + .setTransactionExecutor(executorService) + .build() + + executorService.shutdownNow() + runBlocking { - val executorService = Executors.newCachedThreadPool() - val localDatabase = Room.inMemoryDatabaseBuilder( - ApplicationProvider.getApplicationContext(), TestDatabase::class.java - ) - .setTransactionExecutor(executorService) - .build() - - executorService.shutdownNow() - try { localDatabase.withTransaction { fail("This coroutine should never run.") @@ -801,22 +874,24 @@ .contains("Unable to acquire a thread to perform the database transaction") } } + + executorService.shutdown() + assertThat(executorService.awaitTermination(1, TimeUnit.SECONDS)).isTrue() } @Test fun withTransaction_databaseOpenError() { + val localDatabase = Room.inMemoryDatabaseBuilder( + ApplicationProvider.getApplicationContext(), TestDatabase::class.java + ) + .addCallback(object : RoomDatabase.Callback() { + override fun onOpen(db: SupportSQLiteDatabase) { + // this causes all transaction methods to throw, this can happen IRL + throw RuntimeException("Error opening Database.") + } + }) + .build() runBlocking { - val localDatabase = Room.inMemoryDatabaseBuilder( - ApplicationProvider.getApplicationContext(), TestDatabase::class.java - ) - .addCallback(object : RoomDatabase.Callback() { - override fun onOpen(db: SupportSQLiteDatabase) { - // this causes all transaction methods to throw, this can happen IRL - throw RuntimeException("Error opening Database.") - } - }) - .build() - try { localDatabase.withTransaction { fail("This coroutine should never run.") @@ -830,41 +905,40 @@ @Test fun withTransaction_beginTransaction_error() { - runBlocking { - // delegate and delegate just so that we can throw in beginTransaction() - val localDatabase = Room.inMemoryDatabaseBuilder( - ApplicationProvider.getApplicationContext(), TestDatabase::class.java - ) - .openHelperFactory( - object : SupportSQLiteOpenHelper.Factory { - val factoryDelegate = FrameworkSQLiteOpenHelperFactory() - override fun create( - configuration: SupportSQLiteOpenHelper.Configuration - ): SupportSQLiteOpenHelper { - val helperDelegate = factoryDelegate.create(configuration) - return object : SupportSQLiteOpenHelper by helperDelegate { - override val writableDatabase: SupportSQLiteDatabase - get() { - val databaseDelegate = helperDelegate.writableDatabase - return object : SupportSQLiteDatabase by databaseDelegate { - override fun beginTransaction() { - throw RuntimeException( - "Error beginning transaction." - ) - } - override fun beginTransactionNonExclusive() { - throw RuntimeException( - "Error beginning transaction." - ) - } + // delegate and delegate just so that we can throw in beginTransaction() + val localDatabase = Room.inMemoryDatabaseBuilder( + ApplicationProvider.getApplicationContext(), TestDatabase::class.java + ) + .openHelperFactory( + object : SupportSQLiteOpenHelper.Factory { + val factoryDelegate = FrameworkSQLiteOpenHelperFactory() + override fun create( + configuration: SupportSQLiteOpenHelper.Configuration + ): SupportSQLiteOpenHelper { + val helperDelegate = factoryDelegate.create(configuration) + return object : SupportSQLiteOpenHelper by helperDelegate { + override val writableDatabase: SupportSQLiteDatabase + get() { + val databaseDelegate = helperDelegate.writableDatabase + return object : SupportSQLiteDatabase by databaseDelegate { + override fun beginTransaction() { + throw RuntimeException( + "Error beginning transaction." + ) + } + override fun beginTransactionNonExclusive() { + throw RuntimeException( + "Error beginning transaction." + ) } } - } + } } } - ) - .build() - + } + ) + .build() + runBlocking { try { localDatabase.withTransaction { fail("This coroutine should never run.") @@ -1023,4 +1097,286 @@ assertThat(booksDao.getBooksSuspend()) .contains(addedBook) } + + @Test + fun withTransaction_instantTaskExecutorRule() = runBlocking { + // Not the actual InstantTaskExecutorRule since this test class already uses + // CountingTaskExecutorRule but same behaviour. + ArchTaskExecutor.getInstance().setDelegate(object : TaskExecutor() { + override fun executeOnDiskIO(runnable: Runnable) { + runnable.run() + } + + override fun postToMainThread(runnable: Runnable) { + runnable.run() + } + + override fun isMainThread(): Boolean { + return false + } + }) + database.withTransaction { + booksDao.insertPublisherSuspend( + TestUtil.PUBLISHER.publisherId, + TestUtil.PUBLISHER.name + ) + } + assertThat(booksDao.getPublishers().size).isEqualTo(1) + } + + @Test + fun withTransaction_singleExecutorDispatcher() { + val executor = Executors.newSingleThreadExecutor() + val localDatabase = Room.inMemoryDatabaseBuilder( + ApplicationProvider.getApplicationContext(), TestDatabase::class.java + ) + .setTransactionExecutor(executor) + .build() + runBlocking { + withContext(executor.asCoroutineDispatcher()) { + localDatabase.withTransaction { + localDatabase.booksDao().insertPublisherSuspend( + TestUtil.PUBLISHER.publisherId, + TestUtil.PUBLISHER.name + ) + } + } + } + assertThat(localDatabase.booksDao().getPublishers().size).isEqualTo(1) + + executor.shutdown() + assertThat(executor.awaitTermination(1, TimeUnit.SECONDS)).isTrue() + } + + @Test + fun withTransaction_reentrant_nested() { + val executor = Executors.newSingleThreadExecutor() + val localDatabase = Room.inMemoryDatabaseBuilder( + ApplicationProvider.getApplicationContext(), TestDatabase::class.java + ) + .setTransactionExecutor(executor) + .build() + runBlocking { + withContext(executor.asCoroutineDispatcher()) { + localDatabase.withTransaction { + localDatabase.booksDao().insertPublisherSuspend( + TestUtil.PUBLISHER.publisherId, + TestUtil.PUBLISHER.name + ) + localDatabase.withTransaction { + localDatabase.booksDao().insertBookSuspend(TestUtil.BOOK_1) + } + } + } + } + assertThat(localDatabase.booksDao().getPublishers().size).isEqualTo(1) + assertThat(localDatabase.booksDao().allBooks.size).isEqualTo(1) + + executor.shutdown() + assertThat(executor.awaitTermination(1, TimeUnit.SECONDS)).isTrue() + } + + @Test + fun withTransaction_reentrant_nested_exception() { + val executor = Executors.newSingleThreadExecutor() + val localDatabase = Room.inMemoryDatabaseBuilder( + ApplicationProvider.getApplicationContext(), TestDatabase::class.java + ) + .setTransactionExecutor(executor) + .build() + runBlocking { + withContext(executor.asCoroutineDispatcher()) { + localDatabase.withTransaction { + localDatabase.booksDao().insertPublisherSuspend( + TestUtil.PUBLISHER.publisherId, + TestUtil.PUBLISHER.name + ) + try { + localDatabase.withTransaction { + localDatabase.booksDao().insertBookSuspend(TestUtil.BOOK_1) + throw IOException("Boom!") + } + @Suppress("UNREACHABLE_CODE") + fail("An exception should have been thrown.") + } catch (ex: IOException) { + assertThat(ex).hasMessageThat() + .contains("Boom") + } + } + } + } + assertThat(localDatabase.booksDao().getPublishers()).isEmpty() + assertThat(localDatabase.booksDao().allBooks).isEmpty() + + executor.shutdown() + assertThat(executor.awaitTermination(1, TimeUnit.SECONDS)).isTrue() + } + + @Test + fun withTransaction_reentrant_nested_contextSwitch() { + val executor = Executors.newSingleThreadExecutor() + val localDatabase = Room.inMemoryDatabaseBuilder( + ApplicationProvider.getApplicationContext(), TestDatabase::class.java + ) + .setTransactionExecutor(executor) + .build() + + runBlocking { + withContext(executor.asCoroutineDispatcher()) { + localDatabase.withTransaction { + localDatabase.booksDao().insertPublisherSuspend( + TestUtil.PUBLISHER.publisherId, + TestUtil.PUBLISHER.name + ) + withContext(Dispatchers.IO) { + localDatabase.withTransaction { + localDatabase.booksDao().insertBookSuspend(TestUtil.BOOK_1) + } + } + } + } + } + assertThat(localDatabase.booksDao().getPublishers().size).isEqualTo(1) + assertThat(localDatabase.booksDao().allBooks.size).isEqualTo(1) + + executor.shutdown() + assertThat(executor.awaitTermination(1, TimeUnit.SECONDS)).isTrue() + } + + @Test + fun withTransaction_reentrant_busyExecutor() { + val busyThreadsCount = AtomicInteger() + val executor = + BusyCountingService(busyThreadsCount, Executors.newFixedThreadPool(2)) + val localDatabase = Room.inMemoryDatabaseBuilder( + ApplicationProvider.getApplicationContext(), TestDatabase::class.java + ) + .setTransactionExecutor(executor) + .build() + + // Grab one of the thread and simulate busy work + val busyLatch = CountDownLatch(1) + executor.execute { + busyLatch.await() + } + + runBlocking { + // Using the other thread in the pool this will cause a reentrant situation + withContext(executor.asCoroutineDispatcher()) { + localDatabase.withTransaction { + val transactionThread = Thread.currentThread() + // Suspend transaction thread while freeing the busy thread from the pool + withContext(Dispatchers.IO) { + busyLatch.countDown() + delay(200) + // Only one thread is busy, the transaction thread + assertThat(busyThreadsCount.get()).isEqualTo(1) + } + // Resume in the transaction thread, the recently free thread in the pool that + // is not in a transaction should not be used. + assertThat(Thread.currentThread()).isEqualTo(transactionThread) + localDatabase.booksDao().insertPublisherSuspend( + TestUtil.PUBLISHER.publisherId, + TestUtil.PUBLISHER.name + ) + } + } + } + + assertThat(localDatabase.booksDao().getPublishers().size).isEqualTo(1) + + executor.shutdown() + assertThat(executor.awaitTermination(1, TimeUnit.SECONDS)).isTrue() + } + + @Test + @OptIn(ExperimentalCoroutinesApi::class) + fun withTransaction_runTest() { + runTest { + database.withTransaction { + booksDao.insertPublisherSuspend( + TestUtil.PUBLISHER.publisherId, + TestUtil.PUBLISHER.name + ) + booksDao.insertBookSuspend(TestUtil.BOOK_1.copy(salesCnt = 0)) + booksDao.insertBookSuspend(TestUtil.BOOK_2) + booksDao.deleteUnsoldBooks() + } + assertThat(booksDao.getBooksSuspend()) + .isEqualTo(listOf(TestUtil.BOOK_2)) + } + } + + @Test + fun withTransaction_stress_testMutation() { + val output = mutableListOf<String>() + runBlocking { + repeat(5000) { count -> + database.withTransaction { + output.add("$count") + suspendHere() + output.add("$count") + } + } + } + + val expectedOutput = buildList { + repeat(5000) { count -> + add("$count") + add("$count") + } + } + assertThat(output).isEqualTo(expectedOutput) + } + + @Test + fun withTransaction_stress_dbMutation() { + val context: Context = ApplicationProvider.getApplicationContext() + context.deleteDatabase("test_stress_dbMutation.db") + val db = Room.databaseBuilder( + context, + TestDatabase::class.java, + "test.db" + ).build() + runBlocking { + db.counterDao().upsert(Counter(1, 0)) + repeat(5000) { + launch(Dispatchers.IO) { + db.withTransaction { + val current = db.counterDao().getCounter(1) + suspendHere() + db.counterDao().upsert(current.copy(value = current.value + 1)) + } + } + } + } + runBlocking { + val count = db.counterDao().getCounter(1) + assertThat(count.value).isEqualTo(5000) + } + db.close() + } + + // Utility function to _really_ suspend. + private suspend fun suspendHere(): Unit = suspendCoroutineUninterceptedOrReturn { + it.intercepted().resume(Unit) + COROUTINE_SUSPENDED + } + + // Executor wrapper that counts threads that are busy executing commands. + class BusyCountingService( + val count: AtomicInteger, + val delegate: ExecutorService + ) : ExecutorService by delegate { + override fun execute(command: Runnable) { + delegate.execute { + count.incrementAndGet() + try { + command.run() + } finally { + count.decrementAndGet() + } + } + } + } }
diff --git a/room/integration-tests/kotlintestapp/src/androidTest/java/androidx/room/integration/kotlintestapp/vo/Counter.kt b/room/integration-tests/kotlintestapp/src/androidTest/java/androidx/room/integration/kotlintestapp/vo/Counter.kt new file mode 100644 index 0000000..ce8fd1c --- /dev/null +++ b/room/integration-tests/kotlintestapp/src/androidTest/java/androidx/room/integration/kotlintestapp/vo/Counter.kt
@@ -0,0 +1,26 @@ +/* + * Copyright 2022 The Android Open Source Project + * + * Licensed 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 androidx.room.androidx.room.integration.kotlintestapp.vo + +import androidx.room.Entity +import androidx.room.PrimaryKey + +@Entity +data class Counter( + @PrimaryKey val id: Long, + val value: Int +) \ No newline at end of file
diff --git a/room/room-ktx/src/main/java/androidx/room/RoomDatabaseExt.kt b/room/room-ktx/src/main/java/androidx/room/RoomDatabaseExt.kt index 4edabab..9fc8d5d 100644 --- a/room/room-ktx/src/main/java/androidx/room/RoomDatabaseExt.kt +++ b/room/room-ktx/src/main/java/androidx/room/RoomDatabaseExt.kt
@@ -18,18 +18,17 @@ package androidx.room import androidx.annotation.RestrictTo -import kotlinx.coroutines.Job -import kotlinx.coroutines.asContextElement -import kotlinx.coroutines.runBlocking -import kotlinx.coroutines.suspendCancellableCoroutine -import kotlinx.coroutines.withContext -import java.util.concurrent.Executor import java.util.concurrent.RejectedExecutionException import java.util.concurrent.atomic.AtomicInteger import kotlin.coroutines.ContinuationInterceptor import kotlin.coroutines.CoroutineContext import kotlin.coroutines.coroutineContext import kotlin.coroutines.resume +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.asContextElement +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.suspendCancellableCoroutine +import kotlinx.coroutines.withContext /** * Calls the specified suspending [block] in a database transaction. The transaction will be @@ -43,13 +42,11 @@ * one received by the suspending block. It is recommended that all [Dao] function invoked within * the [block] be suspending functions. * - * The dispatcher used to execute the given [block] will utilize threads from Room's query executor. + * The internal dispatcher used to execute the given [block] will block an utilize a thread from + * Room's transaction executor until the [block] is complete. */ public suspend fun <R> RoomDatabase.withTransaction(block: suspend () -> R): R { - // Use inherited transaction context if available, this allows nested suspending transactions. - val transactionContext = - coroutineContext[TransactionElement]?.transactionDispatcher ?: createTransactionContext() - return withContext(transactionContext) { + val transactionBlock: suspend CoroutineScope.() -> R = transaction@{ val transactionElement = coroutineContext[TransactionElement]!! transactionElement.acquire() try { @@ -59,7 +56,7 @@ val result = block.invoke() @Suppress("DEPRECATION") setTransactionSuccessful() - return@withContext result + return@transaction result } finally { @Suppress("DEPRECATION") endTransaction() @@ -68,6 +65,51 @@ transactionElement.release() } } + // Use inherited transaction context if available, this allows nested suspending transactions. + val transactionDispatcher = coroutineContext[TransactionElement]?.transactionDispatcher + return if (transactionDispatcher != null) { + withContext(transactionDispatcher, transactionBlock) + } else { + startTransactionCoroutine(coroutineContext, transactionBlock) + } +} + +/** + * Suspend caller coroutine and start the transaction coroutine in a thread from the + * [RoomDatabase.transactionExecutor], resuming the caller coroutine with the result once done. + * The [context] will be a parent of the started coroutine to propagating cancellation and release + * the thread when cancelled. + */ +private suspend fun <R> RoomDatabase.startTransactionCoroutine( + context: CoroutineContext, + transactionBlock: suspend CoroutineScope.() -> R +): R = suspendCancellableCoroutine { continuation -> + try { + transactionExecutor.execute { + try { + // Thread acquired, start the transaction coroutine using the parent context. + // The started coroutine will have an event loop dispatcher that we'll use for the + // transaction context. + runBlocking(context.minusKey(ContinuationInterceptor)) { + val dispatcher = coroutineContext[ContinuationInterceptor]!! + val transactionContext = createTransactionContext(dispatcher) + continuation.resume( + withContext(transactionContext, transactionBlock) + ) + } + } catch (ex: Throwable) { + // If anything goes wrong, propagate exception to the calling coroutine. + continuation.cancel(ex) + } + } + } catch (ex: RejectedExecutionException) { + // Couldn't acquire a thread, cancel coroutine. + continuation.cancel( + IllegalStateException( + "Unable to acquire a thread to perform the database transaction.", ex + ) + ) + } } /** @@ -76,8 +118,9 @@ * The context is a combination of a dispatcher, a [TransactionElement] and a thread local element. * * * The dispatcher will dispatch coroutines to a single thread that is taken over from the Room - * query executor. If the coroutine context is switched, suspending DAO functions will be able to - * dispatch to the transaction thread. + * transaction executor. If the coroutine context is switched, suspending DAO functions will be able + * to dispatch to the transaction thread. In reality the dispatcher is the event loop of a + * [runBlocking] started on the dedicated thread. * * * The [TransactionElement] serves as an indicator for inherited context, meaning, if there is a * switch of context, suspending DAO methods will be able to use the indicator to dispatch the @@ -88,61 +131,22 @@ * if a blocking DAO method is invoked within the transaction coroutine. Never assign meaning to * this value, for now all we care is if its present or not. */ -private suspend fun RoomDatabase.createTransactionContext(): CoroutineContext { - val controlJob = Job() - // make sure to tie the control job to this context to avoid blocking the transaction if - // context get cancelled before we can even start using this job. Otherwise, the acquired - // transaction thread will forever wait for the controlJob to be cancelled. - // see b/148181325 - coroutineContext[Job]?.invokeOnCompletion { - controlJob.cancel() - } - val dispatcher = transactionExecutor.acquireTransactionThread(controlJob) - val transactionElement = TransactionElement(controlJob, dispatcher) +private fun RoomDatabase.createTransactionContext( + dispatcher: ContinuationInterceptor +): CoroutineContext { + val transactionElement = TransactionElement(dispatcher) val threadLocalElement = - suspendingTransactionId.asContextElement(System.identityHashCode(controlJob)) + suspendingTransactionId.asContextElement(System.identityHashCode(transactionElement)) return dispatcher + transactionElement + threadLocalElement } /** - * Acquires a thread from the executor and returns a [ContinuationInterceptor] to dispatch - * coroutines to the acquired thread. The [controlJob] is used to control the release of the - * thread by cancelling the job. - */ -private suspend fun Executor.acquireTransactionThread(controlJob: Job): ContinuationInterceptor { - return suspendCancellableCoroutine { continuation -> - continuation.invokeOnCancellation { - // We got cancelled while waiting to acquire a thread, we can't stop our attempt to - // acquire a thread, but we can cancel the controlling job so once it gets acquired it - // is quickly released. - controlJob.cancel() - } - try { - execute { - runBlocking { - // Thread acquired, resume coroutine. - continuation.resume(coroutineContext[ContinuationInterceptor]!!) - controlJob.join() - } - } - } catch (ex: RejectedExecutionException) { - // Couldn't acquire a thread, cancel coroutine. - continuation.cancel( - IllegalStateException( - "Unable to acquire a thread to perform the database transaction.", ex - ) - ) - } - } -} -/** * A [CoroutineContext.Element] that indicates there is an on-going database transaction. * * @hide */ @RestrictTo(RestrictTo.Scope.LIBRARY_GROUP) internal class TransactionElement( - private val transactionThreadControlJob: Job, internal val transactionDispatcher: ContinuationInterceptor ) : CoroutineContext.Element { @@ -153,9 +157,7 @@ /** * Number of transactions (including nested ones) started with this element. - * Call [acquire] to increase the count and [release] to decrease it. If the count reaches zero - * when [release] is invoked then the transaction job is cancelled and the transaction thread - * is released. + * Call [acquire] to increase the count and [release] to decrease it. */ private val referenceCount = AtomicInteger(0) @@ -167,9 +169,6 @@ val count = referenceCount.decrementAndGet() if (count < 0) { throw IllegalStateException("Transaction was never started or was already released.") - } else if (count == 0) { - // Cancel the job that controls the transaction thread, causing it to be released. - transactionThreadControlJob.cancel() } } } \ No newline at end of file
diff --git a/sqlite/sqlite/src/main/java/androidx/sqlite/db/SupportSQLiteDatabase.kt b/sqlite/sqlite/src/main/java/androidx/sqlite/db/SupportSQLiteDatabase.kt index bd253f2cc..820d26a 100644 --- a/sqlite/sqlite/src/main/java/androidx/sqlite/db/SupportSQLiteDatabase.kt +++ b/sqlite/sqlite/src/main/java/androidx/sqlite/db/SupportSQLiteDatabase.kt
@@ -53,8 +53,8 @@ * * ``` * db.beginTransaction() - * try { - * ... + * try { + * ... * db.setTransactionSuccessful() * } finally { * db.endTransaction() @@ -75,10 +75,10 @@ * * ``` * db.beginTransactionNonExclusive() - * try { - * ... + * try { + * ... * db.setTransactionSuccessful() - * } finally { + * } finally { * db.endTransaction() * } * ``` @@ -126,7 +126,7 @@ * db.beginTransactionWithListenerNonExclusive(listener) * try { * ... - * db.setTransactionSuccessful() + * db.setTransactionSuccessful() * } finally { * db.endTransaction() * }