"cannot collect twice from pager" error after configuration change + Room PagingSource change [158784811]

Fixed

Bug

Status Update

No update yet.

Description

ia...@gmail.com

created issue #1

Jun 12, 2020 03:15AM

Component used: Paging Version used: 3.0.0-alpha01

I was moving my app over to Paging 3.0.0-alpha01. It uses Room 2.3.0-alpha01 for support for PagingSource.

Dao:

internal abstract val chosenPhotosPaged: PagingSource<Int, ChosenPhoto>

ViewModel:

internal val chosenPhotos = Pager(PagingConfig(24, 12)) {
    GalleryDatabase.getInstance(application).chosenPhotoDao().chosenPhotosPaged
}.flow.cachedIn(viewModelScope)

UI:

lifecycleScope.launchWhenStarted {
    viewModel.chosenPhotos.collect {
        // Load into a PagingDataAdapter
        chosenPhotosAdapter.submitData(it)
    }
}

Steps to reproduce:

Checkout the paging3 branch of Muzei
Add 30+ images to your device
Install and Activate Muzei
Go to the Sources screen and select the "My Photos" source
Accept the runtime permission
Go to the Settings screen
Hit the FAB at the bottom, then the image button
Select ~2-5 images
Rotate your device
Hit the FAB at the bottom, image, and add additional images

Expected behavior: The new images are seamlessly added

Actual behavior: I received the following exception:

2020-06-11 19:54:18.401 net.nurik.roman.muzei E/AndroidRuntime: FATAL EXCEPTION: main
    Process: net.nurik.roman.muzei, PID: 21268
    java.lang.IllegalStateException: cannot collect twice from pager
        at androidx.paging.PageFetcherSnapshot$pageEventFlow$1.invokeSuspend(PageFetcherSnapshot.kt:94)
        at androidx.paging.PageFetcherSnapshot$pageEventFlow$1.invoke(Unknown Source:10)
        at androidx.paging.CancelableChannelFlowKt$cancelableChannelFlow$1.invokeSuspend(CancelableChannelFlow.kt:35)
        at androidx.paging.CancelableChannelFlowKt$cancelableChannelFlow$1.invoke(Unknown Source:10)
        at kotlinx.coroutines.flow.ChannelFlowBuilder.collectTo$suspendImpl(Builders.kt:327)
        at kotlinx.coroutines.flow.ChannelFlowBuilder.collectTo(Unknown Source:0)
        at kotlinx.coroutines.flow.internal.ChannelFlow$collectToFun$1.invokeSuspend(ChannelFlow.kt:53)
        at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
        at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:56)
        at kotlinx.coroutines.EventLoop.processUnconfinedEvent(EventLoop.common.kt:69)
        at kotlinx.coroutines.DispatchedTaskKt.resumeUnconfined(DispatchedTask.kt:184)
        at kotlinx.coroutines.DispatchedTaskKt.dispatch(DispatchedTask.kt:108)
        at kotlinx.coroutines.CancellableContinuationImpl.dispatchResume(CancellableContinuationImpl.kt:308)
        at kotlinx.coroutines.CancellableContinuationImpl.resumeImpl(CancellableContinuationImpl.kt:318)
        at kotlinx.coroutines.CancellableContinuationImpl.resumeWith(CancellableContinuationImpl.kt:250)
        at kotlinx.coroutines.channels.AbstractChannel$ReceiveElement.resumeReceiveClosed(AbstractChannel.kt:877)
        at kotlinx.coroutines.channels.AbstractSendChannel.helpClose(AbstractChannel.kt:312)
        at kotlinx.coroutines.channels.AbstractSendChannel.close(AbstractChannel.kt:241)
        at kotlinx.coroutines.channels.ChannelCoroutine.close(Unknown Source:2)
        at kotlinx.coroutines.channels.SendChannel$DefaultImpls.close$default(Channel.kt:102)
        at androidx.paging.CancelableChannelFlowKt$cancelableChannelFlow$1$1.invoke(CancelableChannelFlow.kt:33)
        at androidx.paging.CancelableChannelFlowKt$cancelableChannelFlow$1$1.invoke(Unknown Source:2)
        at kotlinx.coroutines.InvokeOnCompletion.invoke(JobSupport.kt:1386)
        at kotlinx.coroutines.JobSupport.notifyCompletion(JobSupport.kt:1529)
        at kotlinx.coroutines.JobSupport.completeStateFinalization(JobSupport.kt:323)
        at kotlinx.coroutines.JobSupport.finalizeFinishingState(JobSupport.kt:240)
        at kotlinx.coroutines.JobSupport.tryMakeCompletingSlowPath(JobSupport.kt:903)
        at kotlinx.coroutines.JobSupport.tryMakeCompleting(JobSupport.kt:860)
        at kotlinx.coroutines.JobSupport.cancelMakeCompleting(JobSupport.kt:693)
        at kotlinx.coroutines.JobSupport.cancelImpl$kotlinx_coroutines_core(JobSupport.kt:664)
        at kotlinx.coroutines.JobSupport.cancelInternal(JobSupport.kt:629)
        at kotlinx.coroutines.JobSupport.cancel(JobSupport.kt:614)
        at kotlinx.coroutines.Job$DefaultImpls.cancel$default(Job.kt:164)
        at androidx.paging.PageFetcherSnapshot.close(PageFetcherSnapshot.kt:183)
        at androidx.paging.PageFetcher$flow$1$2.invokeSuspend(PageFetcher.kt:75)
        at androidx.paging.PageFetcher$flow$1$2.invoke(Unknown Source:18)
        at kotlinx.coroutines.flow.FlowKt__TransformKt$scan$$inlined$unsafeFlow$1$lambda$1.emit(Collect.kt:133)
        at kotlinx.coroutines.flow.FlowKt__ChannelsKt.emitAllImpl$FlowKt__ChannelsKt(Channels.kt:58)
        at kotlinx.coroutines.flow.FlowKt__ChannelsKt$emitAllImpl$1.invokeSuspend(Unknown Source:11)
        at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
        at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:56)
        at android.os.Handler.handleCallback(Handler.java:883)
        at android.os.Handler.dispatchMessage(Handler.java:100)
        at android.os.Looper.loop(Looper.java:214)
        at android.app.ActivityThread.main(ActivityThread.java:7356)
        at java.lang.reflect.Method.invoke(Native Method)
        at com.android.internal.os.RuntimeInit$MethodAndArgsCaller.run(RuntimeInit.java:492)
        at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:930)
2020-06-11 19:54:18.407 net.nurik.roman.muzei I/Process: Sending signal. PID: 21268 SIG: 9

Comments

du...@google.com <du...@google.com> #2Jun 12, 2020 03:35AM

Reassigned to du...@google.com.

Yigit, do you have time to fix it?

reemission of the same liveData is racy

ia...@gmail.com <ia...@gmail.com> #3Jun 12, 2020 03:46AM

yea i'll take it.

va...@gmail.com <va...@gmail.com> #4Jun 13, 2020 03:26PM

Thanks for the detailed analysis. This may not be an issue anymore since we've started using Main.immediate there but I' not sure; I'll try to create a test case.

ap...@google.com <ap...@google.com> #5Jun 15, 2020 03:29PM

just emitting same live data reproduces the issue.

@Test
fun raceTest() {
val subLiveData = MutableLiveData(1)
val subject = liveData(testScope.coroutineContext) {
emitSource(subLiveData)
emitSource(subLiveData) //crashes
}
subject.addObserver().apply {
testScope.advanceUntilIdle()
}
}

yb...@google.com <yb...@google.com> Jun 15, 2020 04:17PM

Marked as fixed.

lu...@gmail.com <lu...@gmail.com> #6Jun 29, 2020 01:24PM

With 2.2.0-alpha04 (that use Main.immediate), the issue seems to be still there (I tested it by calling emitSource() twice, like your test case)

du...@google.com <du...@google.com> #7Jun 29, 2020 03:51PM

yea sorry immediate does not fix it.
I actually have a WIP fix for it:

https://android-review.googlesource.com/c/platform/frameworks/support/+/1112186

if your case is the one i found (emitting same LiveData multiple times, as shown in #5) you can work around it by adding a dummy transformation.

val subLiveData = MutableLiveData(1)
val subject = liveData(testScope.coroutineContext) {
emitSource(subLiveData.map {it })
emitSource(subLiveData.map {it} )
}

ch...@gmail.com <ch...@gmail.com> #8Jul 8, 2020 07:59AM

Project: platform/frameworks/support
Branch: androidx-master-dev

commit af12e75e6b4110f48e44ca121466943909de8f06
Author: Yigit Boyar <yboyar@google.com>
Date: Tue Sep 03 12:58:11 2019

Fix coroutine livedata race condition

This CL fixes a bug in liveData builder where emitting same
LiveData source twice would make it crash because the second
emission registry could possibly happen before first one is
removed as source.

We fix it by using a suspending dispose function. It does feel
a bit hacky but we cannot make DisposableHandle.dispose async
and we do not want to block there. This does not mean that there
is a problem if developer disposes it manually since our emit
functions take care of making sure it disposes (and there is
no other way to add source to the underlying MediatorLiveData)

Bug: 140249349
Test: BuildLiveDataTest#raceTest_*

Change-Id: I0b464c242a583da4669af195cf2504e2adc4de40

M lifecycle/lifecycle-livedata-ktx/api/2.2.0-alpha05.txt
M lifecycle/lifecycle-livedata-ktx/api/current.txt
M lifecycle/lifecycle-livedata-ktx/api/public_plus_experimental_2.2.0-alpha05.txt
M lifecycle/lifecycle-livedata-ktx/api/public_plus_experimental_current.txt
M lifecycle/lifecycle-livedata-ktx/api/restricted_2.2.0-alpha05.txt
M lifecycle/lifecycle-livedata-ktx/api/restricted_current.txt
M lifecycle/lifecycle-livedata-ktx/src/main/java/androidx/lifecycle/CoroutineLiveData.kt
M lifecycle/lifecycle-livedata-ktx/src/test/java/androidx/lifecycle/BuildLiveDataTest.kt

https://android-review.googlesource.com/1112186

https://goto.google.com/android-sha1/af12e75e6b4110f48e44ca121466943909de8f06

du...@google.com <du...@google.com> #9Jul 8, 2020 05:10PM

It's expected to need to call .cachedIn twice there as you've done since re-collection is unsupported on pagingData, operators like combine cause re-collection so without cachedIn you would be expecting to reload the data from scratch any time a new value is emitted to removedItemsFlow

In your example you are correctly caching the filtered data.

After https://android-review.googlesource.com/c/platform/frameworks/support/+/1356724 lands Paging will call the initial .cachedIn for you (presumably in alpha03).

ch...@gmail.com <ch...@gmail.com> #10Jul 10, 2020 06:22AM

In my example i call .cachedIn twice, but actually it doesn't caching the filtered data, so i have to keep the whole remove data and it will filter all remove data in source data every time.

Such as i want to remove data which id is 1， then i remove id 2 ,but when i remove 2,i have to filter 1 again, so i have to save all remove data, so i think the re-collection doesn't work event i call .cachedIn twice

du...@google.com <du...@google.com> #11Jul 11, 2020 12:40AM

In my example i call .cachedIn twice, but actually it doesn't caching the filtered data, so i have to keep the whole remove data and it will filter all remove data in source data every time.

Such as i want to remove data which id is 1， then i remove id 2 ,but when i remove 2,i have to filter 1 again, so i have to save all remove data, so i think the re-collection doesn't work event i call .cachedIn twice

Sorry - I'm not sure I entirely understand what you're doing here. .cachedIn doesn't affect what REFRESH returns, it simply prevents repeated work on recollection within the specified scope (for example if you change from landscape to portrait).

However on each new instance of PagingData due to invalidation, you would still need to filter the entire stream of events, since it's a new stream.

ch...@gmail.com <ch...@gmail.com> #12Jul 11, 2020 02:41AM

I want to do something like notifyItemRemoved , for example there is a collect list ,i want to uncollect the item1, then uncollect the item2, so when i uncollcet item2 i have to filter item1 again ,

I don’t know if you can understand this example, because i think it's useless when call the .cachedIn after filter, because it can't caching the new filter data , so if i want to remove data,i have to keep all item i want to remove even the data already removed.

du...@google.com <du...@google.com> #13Jul 11, 2020 04:34AM

In this case it's recommended to update the backing dataset directly and call invalidate(). We're investigating a page-level flow api that will allow you to control the items themselves without invalidation which you can track in #9.

Calling .cachedIn after .filter prevents your .filter from needing to run again on the same list. For example on configuration change or when navigating between fragments. It has nothing to do with making your filter operation incremental.

ch...@gmail.com <ch...@gmail.com> #14Jul 13, 2020 02:36AM

Thank you for your reply,I hope you can proceed smoothly and have more powerful api.