Merge pull request #7901 from element-hq/feature/bma/mainThreadBlocking

Avoid blocking the main thread in several places
This commit is contained in:
Benoit Marty
2026-10-14 16:39:14 +02:00
committed by GitHub
13 changed files with 117 additions and 74 deletions
@@ -43,6 +43,7 @@ import io.element.android.features.messages.impl.timeline.TimelineController
import io.element.android.features.messages.impl.utils.TextPillificationHelper
import io.element.android.libraries.architecture.AsyncAction
import io.element.android.libraries.architecture.Presenter
import io.element.android.libraries.core.coroutine.CoroutineDispatchers
import io.element.android.libraries.core.extensions.runCatchingExceptions
import io.element.android.libraries.core.mimetype.MimeTypes
import io.element.android.libraries.designsystem.utils.snackbar.SnackbarDispatcher
@@ -110,6 +111,7 @@ import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.merge
import kotlinx.coroutines.flow.stateIn
import kotlinx.coroutines.launch
import kotlinx.coroutines.withContext
import timber.log.Timber
import kotlin.time.Duration.Companion.seconds
import io.element.android.libraries.core.mimetype.MimeTypes.Any as AnyMimeTypes
@@ -145,6 +147,7 @@ class MessageComposerPresenter(
private val featureFlagService: FeatureFlagService,
private val contentScannerService: ContentScannerService,
private val contentValidationCache: EventContentValidationCache,
private val dispatchers: CoroutineDispatchers,
) : Presenter<MessageComposerState> {
@AssistedFactory
interface Factory {
@@ -190,22 +193,22 @@ class MessageComposerPresenter(
.collectAsState(initial = false)
val galleryMediaPicker = mediaPickerProvider.registerGalleryPicker { uri, mimeType ->
handlePickedMedia(uri, mimeType)
localCoroutineScope.handlePickedMedia(uri, mimeType)
}
val galleryMultiMediaPicker = mediaPickerProvider.registerGalleryMultiPicker { uris ->
handlePickedMediaList(uris)
localCoroutineScope.handlePickedMediaList(uris)
}
val filesPicker = mediaPickerProvider.registerFileMultiPicker(AnyMimeTypes) { uris ->
handlePickedMediaList(uris, sendAsFile = true)
localCoroutineScope.handlePickedMediaList(uris, sendAsFile = true)
}
val fileSinglePicker = mediaPickerProvider.registerFilePicker(AnyMimeTypes) { uri, mimeType ->
handlePickedMedia(uri, mimeType ?: MimeTypes.OctetStream, sendAsFile = true)
localCoroutineScope.handlePickedMedia(uri, mimeType ?: MimeTypes.OctetStream, sendAsFile = true)
}
val cameraPhotoPicker = mediaPickerProvider.registerCameraPhotoPicker { uri ->
handlePickedMedia(uri, MimeTypes.Jpeg)
localCoroutineScope.handlePickedMedia(uri, MimeTypes.Jpeg)
}
val cameraVideoPicker = mediaPickerProvider.registerCameraVideoPicker { uri ->
handlePickedMedia(uri, MimeTypes.Mp4)
localCoroutineScope.handlePickedMedia(uri, MimeTypes.Mp4)
}
val isFullScreen = rememberSaveable {
mutableStateOf(false)
@@ -283,17 +286,16 @@ class MessageComposerPresenter(
}
is MessageComposerEvent.SendUri -> {
val inReplyToEventId = (messageComposerContext.composerMode as? MessageComposerMode.Reply)?.eventId
sessionCoroutineScope.sendAttachment(
attachment = Attachment.Media(
localMedia = localMediaFactory.createFromUri(
uri = event.uri,
mimeType = null,
name = null,
formattedFileSize = null
),
),
inReplyToEventId = inReplyToEventId,
)
sessionCoroutineScope.launch {
val localMedia = createLocalMedia(
uri = event.uri,
mimeType = null,
)
sendAttachment(
attachment = Attachment.Media(localMedia),
inReplyToEventId = inReplyToEventId,
)
}
resetComposerModeAfterAttaching()
}
@@ -637,17 +639,28 @@ class MessageComposerPresenter(
}
}
private fun handlePickedMedia(
uri: Uri?,
mimeType: String? = null,
sendAsFile: Boolean = false,
) {
uri ?: return
val localMedia = localMediaFactory.createFromUri(
// Creating the LocalMedia queries the ContentResolver and can decode the media, so do it off the main thread
private suspend fun createLocalMedia(
uri: Uri,
mimeType: String?,
) = withContext(dispatchers.io) {
localMediaFactory.createFromUri(
uri = uri,
mimeType = mimeType,
name = null,
formattedFileSize = null
formattedFileSize = null,
)
}
private fun CoroutineScope.handlePickedMedia(
uri: Uri?,
mimeType: String? = null,
sendAsFile: Boolean = false,
) = launch {
uri ?: return@launch
val localMedia = createLocalMedia(
uri = uri,
mimeType = mimeType,
)
val mediaAttachment = Attachment.Media(localMedia, sendAsFile = sendAsFile)
val inReplyToEventId = (messageComposerContext.composerMode as? MessageComposerMode.Reply)?.eventId
@@ -656,7 +669,7 @@ class MessageComposerPresenter(
resetComposerModeAfterAttaching()
}
private fun handlePickedMediaList(
private fun CoroutineScope.handlePickedMediaList(
uris: List<Uri>,
sendAsFile: Boolean = false,
) {
@@ -665,19 +678,19 @@ class MessageComposerPresenter(
handlePickedMedia(uris.first(), sendAsFile = sendAsFile)
return
}
val attachments = uris.map { uri ->
val localMedia = localMediaFactory.createFromUri(
uri = uri,
mimeType = null,
name = null,
formattedFileSize = null,
)
Attachment.Media(localMedia, sendAsFile = sendAsFile)
}.toImmutableList()
val inReplyToEventId = (messageComposerContext.composerMode as? MessageComposerMode.Reply)?.eventId
navigator.navigateToPreviewAttachments(attachments, inReplyToEventId)
launch {
val attachments = uris.map { uri ->
val localMedia = createLocalMedia(
uri = uri,
mimeType = null,
)
Attachment.Media(localMedia, sendAsFile = sendAsFile)
}.toImmutableList()
val inReplyToEventId = (messageComposerContext.composerMode as? MessageComposerMode.Reply)?.eventId
navigator.navigateToPreviewAttachments(attachments, inReplyToEventId)
resetComposerModeAfterAttaching()
resetComposerModeAfterAttaching()
}
}
private fun resetComposerModeAfterAttaching() {
@@ -324,6 +324,7 @@ class MessageComposerPresenterSlashCommandTest {
featureFlagService = featureFlagService,
contentScannerService = { _, _ -> },
contentValidationCache = InMemoryEventContentValidationCache(),
dispatchers = testCoroutineDispatchers(useUnconfinedTestDispatcher = true),
).apply {
isTesting = true
showTextFormatting = isRichTextEditorEnabled
@@ -1723,6 +1723,7 @@ class MessageComposerPresenterTest : RobolectricTest() {
featureFlagService = featureFlagService,
contentScannerService = { _, _ -> },
contentValidationCache = InMemoryEventContentValidationCache(),
dispatchers = testCoroutineDispatchers(useUnconfinedTestDispatcher = true),
).apply {
isTesting = true
showTextFormatting = isRichTextEditorEnabled
@@ -26,6 +26,7 @@ import io.element.android.libraries.architecture.AsyncAction
import io.element.android.libraries.architecture.AsyncData
import io.element.android.libraries.architecture.Presenter
import io.element.android.libraries.architecture.runUpdatingStateNoSuccess
import io.element.android.libraries.core.coroutine.CoroutineDispatchers
import io.element.android.libraries.core.extensions.runCatchingExceptions
import io.element.android.libraries.di.annotations.SessionCoroutineScope
import io.element.android.libraries.featureflag.api.FeatureFlagService
@@ -57,6 +58,7 @@ import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.launch
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import kotlinx.coroutines.withContext
import timber.log.Timber
import kotlin.time.Duration.Companion.seconds
@@ -76,6 +78,7 @@ class NotificationSettingsPresenter(
@SessionCoroutineScope
private val sessionCoroutineScope: CoroutineScope,
private val featureFlagService: FeatureFlagService,
private val dispatchers: CoroutineDispatchers,
) : Presenter<NotificationSettingsState> {
// Serializes the pick → copy → persist → recreate pipeline per slot. The copier already locks
// its `<slot>.tmp` file, but the persist+recreate window outside the copier is unprotected;
@@ -111,16 +114,19 @@ class NotificationSettingsPresenter(
}
// List of PushProvider -> Distributor
val distributors = remember {
pushService.getAvailablePushProviders()
.flatMap { pushProvider ->
pushProvider.getDistributors().map { distributor ->
pushProvider to distributor
// Getting the distributors can query the PackageManager and the Play Services, so do it off the main thread
val distributors by produceState(emptyList<Pair<PushProvider, Distributor>>()) {
value = withContext(dispatchers.io) {
pushService.getAvailablePushProviders()
.flatMap { pushProvider ->
pushProvider.getDistributors().map { distributor ->
pushProvider to distributor
}
}
}
}
}
// List of Distributors
val availableDistributors = remember {
val availableDistributors = remember(distributors) {
distributors.map { it.second }.toImmutableList()
}
@@ -40,6 +40,7 @@ import io.element.android.tests.testutils.consumeItemsUntilPredicate
import io.element.android.tests.testutils.lambda.lambdaRecorder
import io.element.android.tests.testutils.lambda.value
import io.element.android.tests.testutils.test
import io.element.android.tests.testutils.testCoroutineDispatchers
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.test.TestScope
@@ -1176,6 +1177,7 @@ class NotificationSettingsPresenterTest {
stringProvider = FakeStringProvider(),
sessionCoroutineScope = backgroundScope,
featureFlagService = featureFlagService,
dispatchers = testCoroutineDispatchers(useUnconfinedTestDispatcher = true),
)
}
}
@@ -118,6 +118,7 @@ import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.buffer
import kotlinx.coroutines.flow.distinctUntilChanged
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.flowOn
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.mapNotNull
@@ -320,7 +321,7 @@ class RustMatrixClient(
override fun onUpdate(profile: UserProfile) {
val matrixUser = profile.map()
_userProfile.tryEmit(matrixUser)
sessionCoroutineScope.launch {
sessionCoroutineScope.launch(sessionDispatcher) {
storeUserProfile(matrixUser)
}
}
@@ -1013,6 +1014,8 @@ class RustMatrixClient(
searchBackfillService.cancelPeriodicSearchBackfill()
}
}
// Checking the pending work is a blocking call, do not do it on the main thread
.flowOn(sessionDispatcher)
.launchIn(sessionCoroutineScope)
}
}
@@ -158,6 +158,7 @@ class JoinedRustRoom(
inner = innerRoom.threadListService(),
contentMapper = TimelineEventContentMapper(),
roomCoroutineScope = roomCoroutineScope,
roomDispatcher = roomDispatcher,
)
override val syncUpdateFlow = flow {
@@ -17,10 +17,12 @@ import io.element.android.libraries.matrix.api.room.threads.ThreadsListService
import io.element.android.libraries.matrix.impl.timeline.item.event.TimelineEventContentMapper
import io.element.android.libraries.matrix.impl.timeline.item.event.map
import io.element.android.libraries.matrix.impl.util.mxCallbackFlow
import kotlinx.coroutines.CoroutineDispatcher
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Job
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.flowOn
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.flow.onStart
@@ -33,6 +35,7 @@ import org.matrix.rustcomponents.sdk.ThreadListService as InnerThreadListService
class RustThreadsListService(
private val inner: InnerThreadListService,
private val roomCoroutineScope: CoroutineScope,
private val roomDispatcher: CoroutineDispatcher,
private val contentMapper: TimelineEventContentMapper = TimelineEventContentMapper(),
) : ThreadsListService {
private var itemSubscriptionJob: Job? = null
@@ -63,6 +66,8 @@ class RustThreadsListService(
updated.apply(diff, contentMapper)
items.value = updated
}
// Fetching and mapping the items can be costly, do not do it on the main thread
.flowOn(roomDispatcher)
.launchIn(roomCoroutineScope)
}
@@ -33,7 +33,9 @@ import io.element.android.services.analytics.test.FakeAnalyticsService
import io.element.android.services.toolbox.test.systemclock.FakeSystemClock
import io.element.android.tests.testutils.lambda.lambdaRecorder
import io.element.android.tests.testutils.testCoroutineDispatchers
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.test.TestScope
import kotlinx.coroutines.test.runCurrent
import kotlinx.coroutines.test.runTest
import org.junit.Test
import org.matrix.rustcomponents.sdk.Client
@@ -41,6 +43,7 @@ import org.matrix.rustcomponents.sdk.ClientDelegate
import java.io.File
import org.matrix.rustcomponents.sdk.SyncService as ClientSyncService
@OptIn(ExperimentalCoroutinesApi::class)
class RustMatrixClientFactoryTest {
@Test
fun test() = runTest {
@@ -51,6 +54,7 @@ class RustMatrixClientFactoryTest {
val result = sut.create(aSessionData())
assertThat(result.sessionId).isEqualTo(SessionId("@alice:server.org"))
runCurrent()
// 2 workers are scheduled: one for the search backfill and one for the vacuum
scheduleWorkersLambda.assertions().isCalledExactly(2)
result.destroy()
@@ -18,6 +18,7 @@ import io.element.android.libraries.matrix.test.A_TIMESTAMP
import io.element.android.libraries.matrix.test.A_USER_ID
import io.element.android.tests.testutils.lambda.lambdaRecorder
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.test.StandardTestDispatcher
import kotlinx.coroutines.test.TestScope
import kotlinx.coroutines.test.runCurrent
import kotlinx.coroutines.test.runTest
@@ -122,6 +123,7 @@ class RustThreadsListServiceTest {
) = RustThreadsListService(
inner = inner,
roomCoroutineScope = backgroundScope,
roomDispatcher = StandardTestDispatcher(testScheduler),
)
private fun aRustThreadListUpdate() = ThreadListUpdate.Append(
@@ -14,6 +14,7 @@ import android.os.PowerManager
import androidx.core.content.getSystemService
import dev.zacsweers.metro.AppScope
import dev.zacsweers.metro.ContributesBinding
import io.element.android.libraries.core.coroutine.CoroutineDispatchers
import io.element.android.libraries.core.extensions.runCatchingExceptions
import io.element.android.libraries.di.annotations.ApplicationContext
import io.element.android.libraries.matrix.api.core.EventId
@@ -23,23 +24,25 @@ import io.element.android.libraries.push.impl.PushDatabase
import io.element.android.libraries.push.impl.db.PushHistory
import io.element.android.libraries.push.impl.db.PushRequest
import io.element.android.services.toolbox.api.systemclock.SystemClock
import kotlinx.coroutines.withContext
import kotlin.time.Instant
@ContributesBinding(AppScope::class)
class DefaultPushHistoryService(
private val pushDatabase: PushDatabase,
private val systemClock: SystemClock,
private val dispatchers: CoroutineDispatchers,
@ApplicationContext context: Context,
) : PushHistoryService {
private val powerManager = context.getSystemService<PowerManager>()
private val packageName = context.packageName
override suspend fun insertOrUpdatePushRequest(pushRequest: PushRequest): Result<Unit> {
return runCatchingExceptions { pushDatabase.pushRequestQueries.insertPushRequest(pushRequest).await() }
override suspend fun insertOrUpdatePushRequest(pushRequest: PushRequest): Result<Unit> = withContext(dispatchers.io) {
runCatchingExceptions<Unit> { pushDatabase.pushRequestQueries.insertPushRequest(pushRequest).await() }
}
override suspend fun insertOrUpdatePushRequests(pushRequests: List<PushRequest>): Result<Unit> {
return runCatchingExceptions {
override suspend fun insertOrUpdatePushRequests(pushRequests: List<PushRequest>): Result<Unit> = withContext(dispatchers.io) {
runCatchingExceptions {
pushDatabase.transaction {
for (request in pushRequests) {
pushDatabase.pushRequestQueries.insertPushRequest(request)
@@ -48,8 +51,8 @@ class DefaultPushHistoryService(
}
}
override suspend fun getPendingPushRequests(sessionId: SessionId, since: Instant?): Result<List<PushRequest>> {
return runCatchingExceptions {
override suspend fun getPendingPushRequests(sessionId: SessionId, since: Instant?): Result<List<PushRequest>> = withContext(dispatchers.io) {
runCatchingExceptions {
pushDatabase.transactionWithResult {
val sinceTimeMillis = since?.toEpochMilliseconds() ?: 0
pushDatabase.pushRequestQueries.selectAllPendingForSession(sessionId.value, sinceTimeMillis).executeAsList()
@@ -57,14 +60,14 @@ class DefaultPushHistoryService(
}
}
override suspend fun removeOldPushRequests(sessionId: SessionId): Result<Unit> {
return runCatchingExceptions {
override suspend fun removeOldPushRequests(sessionId: SessionId): Result<Unit> = withContext(dispatchers.io) {
runCatchingExceptions<Unit> {
val keepAmount = 100L
pushDatabase.pushRequestQueries.removeOldest(keepAmount)
}
}
override fun onPushResult(
override suspend fun onPushResult(
providerInfo: String,
eventId: EventId?,
roomId: RoomId?,
@@ -91,19 +94,21 @@ class DefaultPushHistoryService(
}
}.takeIf { it.isNotEmpty() }
pushDatabase.pushHistoryQueries.insertPushHistory(
PushHistory(
pushDate = systemClock.epochMillis(),
providerInfo = providerInfo,
eventId = eventId?.value,
roomId = roomId?.value,
sessionId = sessionId?.value,
hasBeenResolved = if (hasBeenResolved) 1 else 0,
comment = finalComment,
withContext(dispatchers.io) {
pushDatabase.pushHistoryQueries.insertPushHistory(
PushHistory(
pushDate = systemClock.epochMillis(),
providerInfo = providerInfo,
eventId = eventId?.value,
roomId = roomId?.value,
sessionId = sessionId?.value,
hasBeenResolved = if (hasBeenResolved) 1 else 0,
comment = finalComment,
)
)
)
// Keep only the last 1_000 events
pushDatabase.pushHistoryQueries.removeOldest(1_000)
// Keep only the last 1_000 events
pushDatabase.pushHistoryQueries.removeOldest(1_000)
}
}
}
@@ -20,7 +20,7 @@ interface PushHistoryService {
* Create a new push history entry.
* Do not use directly, prefer using the extension functions.
*/
fun onPushResult(
suspend fun onPushResult(
providerInfo: String,
eventId: EventId?,
roomId: RoomId?,
@@ -52,7 +52,7 @@ interface PushHistoryService {
suspend fun removeOldPushRequests(sessionId: SessionId): Result<Unit>
}
fun PushHistoryService.onInvalidPushReceived(
suspend fun PushHistoryService.onInvalidPushReceived(
providerInfo: String,
data: String,
) = onPushResult(
@@ -65,7 +65,7 @@ fun PushHistoryService.onInvalidPushReceived(
comment = "Invalid or ignored push data:\n$data",
)
fun PushHistoryService.onUnableToRetrieveSession(
suspend fun PushHistoryService.onUnableToRetrieveSession(
providerInfo: String,
eventId: EventId,
roomId: RoomId,
@@ -80,7 +80,7 @@ fun PushHistoryService.onUnableToRetrieveSession(
comment = "Unable to retrieve session: $reason",
)
fun PushHistoryService.onUnableToResolveEvent(
suspend fun PushHistoryService.onUnableToResolveEvent(
providerInfo: String,
eventId: EventId,
roomId: RoomId,
@@ -96,7 +96,7 @@ fun PushHistoryService.onUnableToResolveEvent(
comment = "Unable to resolve event: $reason",
)
fun PushHistoryService.onSuccess(
suspend fun PushHistoryService.onSuccess(
providerInfo: String,
eventId: EventId,
roomId: RoomId,
@@ -117,7 +117,7 @@ fun PushHistoryService.onSuccess(
},
)
fun PushHistoryService.onDiagnosticPush(
suspend fun PushHistoryService.onDiagnosticPush(
providerInfo: String,
) = onPushResult(
providerInfo = providerInfo,
@@ -30,7 +30,7 @@ class FakePushHistoryService(
private val getPendingPushRequests: (SessionId, Instant?) -> Result<List<PushRequest>> = { _, _ -> lambdaError() },
private val removeOldPushRequests: (SessionId) -> Result<Unit> = { lambdaError() },
) : PushHistoryService {
override fun onPushResult(
override suspend fun onPushResult(
providerInfo: String,
eventId: EventId?,
roomId: RoomId?,