From a9b1242686aff04f4b3d06f03f29379642fbfd26 Mon Sep 17 00:00:00 2001 From: eliotcougar Date: Sun, 13 Sep 2026 04:51:57 +0300 Subject: [PATCH] Coalesce bulk probe result updates (#6205) --- .../java/com/v2ray/ang/dto/RealPingResult.kt | 9 ++ .../com/v2ray/ang/service/CoreTestService.kt | 8 +- .../com/v2ray/ang/ui/main/MainRepository.kt | 26 ++--- .../com/v2ray/ang/ui/main/MainServiceEvent.kt | 3 +- .../com/v2ray/ang/ui/main/MainViewModel.kt | 103 ++++++++++++++++-- 5 files changed, 125 insertions(+), 24 deletions(-) create mode 100644 V2rayNG/app/src/main/java/com/v2ray/ang/dto/RealPingResult.kt diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/dto/RealPingResult.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/dto/RealPingResult.kt new file mode 100644 index 000000000..def53784c --- /dev/null +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/dto/RealPingResult.kt @@ -0,0 +1,9 @@ +package com.v2ray.ang.dto + +import java.io.Serializable + +/** One persisted delay-test result delivered from the task process to the UI. */ +data class RealPingResult( + val guid: String, + val delayMillis: Long, +) : Serializable diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/service/CoreTestService.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/service/CoreTestService.kt index 19726b644..3ba2e7219 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/service/CoreTestService.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/service/CoreTestService.kt @@ -10,6 +10,7 @@ import com.v2ray.ang.AppConfig import com.v2ray.ang.R import com.v2ray.ang.core.CoreNativeManager import com.v2ray.ang.dto.RealPingEvent +import com.v2ray.ang.dto.RealPingResult import com.v2ray.ang.dto.TestServiceMessage import com.v2ray.ang.enums.NotificationChannelType import com.v2ray.ang.extension.serializable @@ -151,7 +152,12 @@ class CoreTestService : Service() { is RealPingEvent.Result -> { MmkvManager.encodeServerTestDelayMillis(event.guid, event.delayMillis) - MessageHelper.sendMsg2UI(this, AppConfig.MSG_MEASURE_CONFIG_SUCCESS, event.guid, requestId) + MessageHelper.sendMsg2UI( + this, + AppConfig.MSG_MEASURE_CONFIG_SUCCESS, + RealPingResult(event.guid, event.delayMillis), + requestId, + ) } is RealPingEvent.Finish -> { diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainRepository.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainRepository.kt index 4a38f282e..cf00177f4 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainRepository.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainRepository.kt @@ -9,6 +9,7 @@ import com.v2ray.ang.AngApplication import com.v2ray.ang.AppConfig import com.v2ray.ang.R import com.v2ray.ang.dto.ConnectionTestResult +import com.v2ray.ang.dto.RealPingResult import com.v2ray.ang.dto.SubscriptionUpdateResult import com.v2ray.ang.dto.TestServiceMessage import com.v2ray.ang.dto.entities.ProfileItem @@ -24,10 +25,9 @@ import com.v2ray.ang.handler.SubscriptionUpdater import com.v2ray.ang.helper.MessageHelper import com.v2ray.ang.util.LogUtil import com.v2ray.ang.util.Utils -import kotlinx.coroutines.channels.BufferOverflow -import kotlinx.coroutines.flow.MutableSharedFlow -import kotlinx.coroutines.flow.SharedFlow -import kotlinx.coroutines.flow.asSharedFlow +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.receiveAsFlow import java.util.concurrent.atomic.AtomicBoolean class MainRepository( @@ -39,13 +39,10 @@ class MainRepository( private val closed = AtomicBoolean(false) - private val _mainServiceEvent = MutableSharedFlow( - replay = 0, - extraBufferCapacity = 64, - onBufferOverflow = BufferOverflow.DROP_OLDEST - ) + // Probe results are finite and must remain lossless until the ViewModel coalesces them. + private val mainServiceEventChannel = Channel(Channel.UNLIMITED) - override val mainServiceEvent: SharedFlow = _mainServiceEvent.asSharedFlow() + override val mainServiceEvent: Flow = mainServiceEventChannel.receiveAsFlow() private val serviceReceiver = object : BroadcastReceiver() { override fun onReceive(context: Context?, intent: Intent?) { @@ -63,7 +60,9 @@ class MainRepository( ?.let { MainServiceEvent.MeasureDelayResult(it, requestId) } AppConfig.MSG_MEASURE_DELAY_CANCEL -> MainServiceEvent.MeasureDelayCancelled(requestId) - AppConfig.MSG_MEASURE_CONFIG_SUCCESS -> MainServiceEvent.MeasureConfigSuccess(requestId) + AppConfig.MSG_MEASURE_CONFIG_SUCCESS -> safeIntent + .serializable("content") + ?.let { MainServiceEvent.MeasureConfigSuccess(it, requestId) } AppConfig.MSG_MEASURE_CONFIG_NOTIFY -> MainServiceEvent.MeasureConfigNotify( safeIntent.getStringExtra("content").orEmpty(), requestId ) @@ -75,7 +74,7 @@ class MainRepository( else -> null } - event?.let { _mainServiceEvent.tryEmit(it) } + event?.let { mainServiceEventChannel.trySend(it) } } } @@ -101,6 +100,7 @@ class MainRepository( }.onFailure { LogUtil.e(AppConfig.TAG, "Failed to unregister main service receiver", it) } + mainServiceEventChannel.close() } override fun getSelectedSubscriptionId(): String = @@ -219,7 +219,7 @@ class MainRepository( override fun testCurrentServerRealPing(requestId: String) { MessageHelper.sendMsg2ServiceForResult(app, AppConfig.MSG_MEASURE_DELAY, requestId) { handled -> - if (!handled) _mainServiceEvent.tryEmit(MainServiceEvent.MeasureDelayCancelled(requestId)) + if (!handled) mainServiceEventChannel.trySend(MainServiceEvent.MeasureDelayCancelled(requestId)) } } diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainServiceEvent.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainServiceEvent.kt index 1e27d7ab5..376422803 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainServiceEvent.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainServiceEvent.kt @@ -1,6 +1,7 @@ package com.v2ray.ang.ui.main import com.v2ray.ang.dto.ConnectionTestResult +import com.v2ray.ang.dto.RealPingResult sealed class MainServiceEvent { data object StateRunning : MainServiceEvent() @@ -10,7 +11,7 @@ sealed class MainServiceEvent { data object StateStopSuccess : MainServiceEvent() data class MeasureDelayResult(val result: ConnectionTestResult, val requestId: String) : MainServiceEvent() data class MeasureDelayCancelled(val requestId: String) : MainServiceEvent() - data class MeasureConfigSuccess(val requestId: String) : MainServiceEvent() + data class MeasureConfigSuccess(val result: RealPingResult, val requestId: String) : MainServiceEvent() data class MeasureConfigNotify(val progress: String, val requestId: String) : MainServiceEvent() data class MeasureConfigFinish(val requestId: String) : MainServiceEvent() data class MeasureConfigCancelled(val requestId: String) : MainServiceEvent() diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainViewModel.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainViewModel.kt index 90eb076d3..75bc0f02c 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainViewModel.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainViewModel.kt @@ -9,6 +9,7 @@ import com.v2ray.ang.R import com.v2ray.ang.dto.ConnectionTestResult import com.v2ray.ang.dto.GroupMapItem import com.v2ray.ang.dto.LocateTarget +import com.v2ray.ang.dto.RealPingResult import com.v2ray.ang.dto.TestServiceMessage import com.v2ray.ang.dto.entities.ProfileItem import com.v2ray.ang.dto.entities.ServersCache @@ -24,6 +25,7 @@ import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.CoroutineDispatcher import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job +import kotlinx.coroutines.cancelAndJoin import kotlinx.coroutines.currentCoroutineContext import kotlinx.coroutines.ensureActive import kotlinx.coroutines.flow.MutableStateFlow @@ -40,6 +42,30 @@ import kotlinx.coroutines.withContext import java.util.concurrent.ConcurrentHashMap import java.util.regex.PatternSyntaxException +private fun applyTestDelayResults( + servers: List, + updates: Map, +): List = servers.map { server -> + val delayMillis = updates[server.guid] + if (delayMillis == null || delayMillis == server.testDelayMillis) { + server + } else { + server.copy(testDelayMillis = delayMillis) + } +} + +private fun applyTestDelayResultsToRows( + rows: List, + updates: Map, +): List = rows.map { row -> + val delayMillis = updates[row.guid] + if (delayMillis == null || delayMillis == row.testDelayMillis) { + row + } else { + row.copy(testDelayMillis = delayMillis) + } +} + class MainViewModel( application: Application, private val dataSource: MainDataSource @@ -77,6 +103,8 @@ class MainViewModel( private var preloadJob: Job? = null private var selectedGroupLoadJob: Job? = null private var reloadJob: Job? = null + private var testResultFlushJob: Job? = null + private val pendingTestResults = linkedMapOf() private val testRequests = MainTestRequests() private var bulkTestJob: Job? = null @@ -119,11 +147,7 @@ class MainViewModel( is MainServiceEvent.MeasureConfigSuccess -> { val request = testRequests.bulk?.takeIf { it.id == event.requestId } ?: return - viewModelScope.launch(ioDispatcher) { - val gid = request.groupId - cacheMutex.withLock { groupDataCache.remove(gid) } - updateGroupUi(gid, loadGroup(gid, forceRefresh = true)) - } + queueTestResult(event.result, request) } is MainServiceEvent.MeasureConfigNotify -> { @@ -133,7 +157,14 @@ class MainViewModel( } is MainServiceEvent.MeasureConfigFinish -> { - onTestsFinished(event.requestId) + val request = testRequests.bulk?.takeIf { it.id == event.requestId } ?: return + val scheduledFlush = testResultFlushJob + testResultFlushJob = viewModelScope.launch { + scheduledFlush?.cancelAndJoin() + if (testRequests.bulk?.id != request.id) return@launch + flushPendingTestResults(request) + onTestsFinished(request.id) + } } is MainServiceEvent.MeasureDelayCancelled -> { @@ -141,11 +172,48 @@ class MainViewModel( } is MainServiceEvent.MeasureConfigCancelled -> { - if (testRequests.completeBulk(event.requestId) != null) resetTestStatus() + if (testRequests.completeBulk(event.requestId) != null) { + cancelPendingTestResults() + resetTestStatus() + } } } } + private fun queueTestResult(result: RealPingResult, request: MainTestRequests.Bulk) { + pendingTestResults[result.guid] = result.delayMillis + if (testResultFlushJob?.isActive == true) return + + testResultFlushJob = viewModelScope.launch { + while (pendingTestResults.isNotEmpty()) { + delay(TEST_RESULT_FLUSH_INTERVAL_MS) + flushPendingTestResults(request) + } + } + } + + private suspend fun flushPendingTestResults(request: MainTestRequests.Bulk) { + if (pendingTestResults.isEmpty()) return + if (testRequests.bulk?.id != request.id) return + + val updates = cacheMutex.withLock { + val drained = pendingTestResults.toMap() + pendingTestResults.clear() + groupDataCache[request.groupId]?.let { cached -> + groupDataCache[request.groupId] = applyTestDelayResults(cached, drained) + } + drained + } + if (updates.isEmpty()) return + if (testRequests.bulk?.id != request.id) return + mutableServerGroupState(request.groupId).update { current -> + current.copy( + servers = applyTestDelayResults(current.servers, updates), + rows = applyTestDelayResultsToRows(current.rows, updates), + ) + } + } + internal fun formatStatus(status: MainStatus): String = when (status) { MainStatus.Disconnected -> dataSource.getString(R.string.connection_not_connected) MainStatus.Connected -> dataSource.getString(R.string.connection_connected) @@ -740,6 +808,7 @@ class MainViewModel( bulkTestJob = null testRequests.cancelBulk() testRequests.invalidateCurrent() + cancelPendingTestResults() resetTestStatus() dataSource.cancelAllPing() } @@ -789,15 +858,27 @@ class MainViewModel( } bulkTestJob = viewModelScope.launch { withContext(ioDispatcher) { + dataSource.clearAllTestDelayResults(serverGuids) + val resetGuids = serverGuids.toHashSet() cacheMutex.withLock { - dataSource.clearAllTestDelayResults(serverGuids) - groupDataCache.remove(groupId) + groupDataCache[groupId]?.let { cached -> + groupDataCache[groupId] = cached.map { server -> + if (server.guid !in resetGuids || server.testDelayMillis == 0L) server + else server.copy(testDelayMillis = 0L) + } + } } } dataSource.sendMsg2TestService(message, request.id) } } + private fun cancelPendingTestResults() { + testResultFlushJob?.cancel() + testResultFlushJob = null + pendingTestResults.clear() + } + fun testCurrentServerRealPing() { if (!uiState.value.isRunning) return val requestId = testRequests.beginCurrent() @@ -869,4 +950,8 @@ class MainViewModel( throw IllegalArgumentException("Unknown ViewModel class") } } + + private companion object { + const val TEST_RESULT_FLUSH_INTERVAL_MS = 500L + } }