From 67ed1d1367fe7a9c56672e4a0f1e4cfd8bef7326 Mon Sep 17 00:00:00 2001 From: Eliot the Cougar Date: Sat, 1 Aug 2026 20:37:02 +0300 Subject: [PATCH 01/10] Probe policy groups progressively in an isolated process Merge one delay-test batch into a disposable Xray process, honor the configured profile concurrency, and update each result as its observatory candidates finish. --- V2rayNG/app/src/main/AndroidManifest.xml | 2 +- .../com/v2ray/ang/core/CoreConfigManager.kt | 31 ++ .../com/v2ray/ang/core/CoreNativeManager.kt | 6 +- .../ang/core/OutboundProbeConfigBuilder.kt | 291 ++++++++++++++++++ .../com/v2ray/ang/dto/OutboundProbePlan.kt | 14 + .../com/v2ray/ang/handler/SettingsManager.kt | 2 +- .../com/v2ray/ang/service/CoreTestService.kt | 138 ++++----- .../ang/service/RealPingWorkerService.kt | 183 ++++++----- .../com/v2ray/ang/ui/main/MainBottomBar.kt | 3 +- .../java/com/v2ray/ang/ui/main/MainScreen.kt | 1 + .../com/v2ray/ang/ui/main/MainViewModel.kt | 3 +- 11 files changed, 517 insertions(+), 157 deletions(-) create mode 100644 V2rayNG/app/src/main/java/com/v2ray/ang/core/OutboundProbeConfigBuilder.kt create mode 100644 V2rayNG/app/src/main/java/com/v2ray/ang/dto/OutboundProbePlan.kt diff --git a/V2rayNG/app/src/main/AndroidManifest.xml b/V2rayNG/app/src/main/AndroidManifest.xml index 0238a58610..bfc145632d 100644 --- a/V2rayNG/app/src/main/AndroidManifest.xml +++ b/V2rayNG/app/src/main/AndroidManifest.xml @@ -256,7 +256,7 @@ android:name=".service.CoreTestService" android:exported="false" android:foregroundServiceType="specialUse" - android:process=":RunSoLibV2RayDaemon"> + android:process=":OutboundProbe"> diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt index e40ce604c6..1da3ad433e 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt @@ -7,6 +7,7 @@ import com.google.gson.JsonObject import com.v2ray.ang.AppConfig import com.v2ray.ang.dto.ConfigResult import com.v2ray.ang.dto.CoreConfigContext +import com.v2ray.ang.dto.OutboundProbePlan import com.v2ray.ang.dto.V2rayConfig import com.v2ray.ang.dto.entities.ProfileItem import com.v2ray.ang.dto.entities.RulesetItem @@ -83,6 +84,25 @@ object CoreConfigManager { } } + /** Builds one isolated Xray configuration for a complete UI delay-test batch. */ + fun getV2rayConfig4BatchSpeedtest(context: Context, guids: List): OutboundProbePlan { + val sources = mutableListOf() + val failedGuids = mutableListOf() + guids.distinct().forEach { guid -> + val result = getV2rayConfig4Speedtest(context, guid) + if (result.status && result.content.isNotBlank()) { + sources += OutboundProbeConfigBuilder.Source(guid, result.content) + } else { + failedGuids += guid + } + } + val plan = OutboundProbeConfigBuilder.build( + sources = sources, + destination = SettingsManager.getDelayTestUrl(), + ) + return plan.copy(failedGuids = (failedGuids + plan.failedGuids).distinct()) + } + /** * Build configuration for custom profiles. */ @@ -435,7 +455,18 @@ object CoreConfigManager { private fun postProcessForSpeedtest(v2rayConfig: V2rayConfig) { v2rayConfig.log.loglevel = MmkvManager.decodeSettingsString(AppConfig.PREF_LOGLEVEL) ?: "warning" v2rayConfig.inbounds.clear() + val usesPrimaryBalancer = v2rayConfig.routing.balancers + ?.any { it.tag == AppConfig.TAG_BALANCER } + ?: false v2rayConfig.routing.rules.clear() + if (usesPrimaryBalancer) { + v2rayConfig.routing.rules.add( + V2rayConfig.RoutingBean.RulesBean( + network = "tcp,udp", + balancerTag = AppConfig.TAG_BALANCER, + ) + ) + } v2rayConfig.dns = null v2rayConfig.fakedns = null v2rayConfig.stats = null diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreNativeManager.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreNativeManager.kt index 663d0629d4..ff31ee090b 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreNativeManager.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreNativeManager.kt @@ -8,6 +8,7 @@ import go.Seq import libv2ray.CoreCallbackHandler import libv2ray.CoreController import libv2ray.Libv2ray +import libv2ray.OutboundProbeController import java.util.concurrent.atomic.AtomicBoolean /** @@ -67,6 +68,9 @@ object CoreNativeManager { } } + fun newOutboundProbeController(): OutboundProbeController = + Libv2ray.newOutboundProbeController() + /** * Measure outbound connection delay. * @@ -97,4 +101,4 @@ object CoreNativeManager { throw e } } -} \ No newline at end of file +} diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/core/OutboundProbeConfigBuilder.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/core/OutboundProbeConfigBuilder.kt new file mode 100644 index 0000000000..91780b25cb --- /dev/null +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/core/OutboundProbeConfigBuilder.kt @@ -0,0 +1,291 @@ + +package com.v2ray.ang.core + +import com.google.gson.JsonArray +import com.google.gson.JsonObject +import com.v2ray.ang.dto.OutboundProbePlan +import com.v2ray.ang.dto.OutboundProbeProfilePlan +import com.v2ray.ang.util.JsonUtil + +/** + * Combines independently generated speed-test configurations into one core. + * + * Every source gets a private tag namespace. Only outbound definitions and the + * primary policy-group balancer are retained. AndroidLib drives the unchanged + * upstream BurstObservatory.Check API with profile-level concurrency and + * progressive group results, so unrelated routing and inbound state must + * not affect another profile in the same batch. + */ +object OutboundProbeConfigBuilder { + data class Source(val guid: String, val content: String) + + fun build( + sources: List, + destination: String, + timeout: String = DEFAULT_PROBE_TIMEOUT, + samples: Int = 1, + ): OutboundProbePlan { + require(destination.isNotBlank()) { "probe destination is empty" } + require(samples > 0) { "probe sample count must be positive" } + + val mergedOutbounds = JsonArray() + val mergedBalancers = JsonArray() + val profiles = mutableListOf() + val failedGuids = mutableListOf() + var batchSamples = samples + var batchTimeout = timeout + var leastLoadHttpMethod: String? = null + + sources.distinctBy { it.guid }.forEachIndexed { index, source -> + val root = JsonUtil.parseString(source.content) + val outbounds = root?.array("outbounds") + if (outbounds == null || outbounds.size() == 0) { + failedGuids += source.guid + return@forEachIndexed + } + + val namespace = "probe-$index-" + val outboundElements = outbounds.toList() + if (outboundElements.any { !it.isJsonObject }) { + failedGuids += source.guid + return@forEachIndexed + } + val outboundObjects = outboundElements.map { it.asJsonObject.deepCopy() } + val originalTags = outboundObjects.mapIndexed { outboundIndex, outbound -> + outbound.string("tag") ?: "outbound-$outboundIndex" + } + if (originalTags.toSet().size != originalTags.size) { + failedGuids += source.guid + return@forEachIndexed + } + + val tagMap = linkedMapOf() + outboundObjects.forEachIndexed { outboundIndex, outbound -> + val originalTag = originalTags[outboundIndex] + val probeTag = "$namespace$originalTag" + tagMap[originalTag] = probeTag + outbound.addProperty("tag", probeTag) + } + if (outboundObjects.any { !remapOutboundReferences(it, tagMap) }) { + failedGuids += source.guid + return@forEachIndexed + } + + val routing = root.obj("routing") + val routingRules = routing?.array("rules") + ?.mapNotNull { it.takeIf { rule -> rule.isJsonObject }?.asJsonObject } + .orEmpty() + val primaryBalancerTag = routingRules + .lastOrNull { it.isCatchAllRule() && it.string("balancerTag") != null } + ?.string("balancerTag") + + val originalBalancer = primaryBalancerTag?.let { wanted -> + routing?.array("balancers") + ?.mapNotNull { it.takeIf { balancer -> balancer.isJsonObject }?.asJsonObject } + ?.firstOrNull { it.string("tag") == wanted } + } + if ((primaryBalancerTag != null && originalBalancer == null) || + (primaryBalancerTag == null && routingRules.any { it.string("balancerTag") != null }) + ) { + // Direct batch probing cannot reproduce conditional routing. + // Accept only the catch-all policy-group form emitted by v2rayNG. + failedGuids += source.guid + return@forEachIndexed + } + + val localBalancers = JsonArray() + val profile = if (originalBalancer != null) { + buildPolicyProfile( + source.guid, + namespace, + originalBalancer, + tagMap, + localBalancers, + ) + } else { + val catchAllOutbound = routingRules.lastOrNull { + it.isCatchAllRule() && it.string("outboundTag") != null + }?.string("outboundTag") + if (catchAllOutbound != null && catchAllOutbound !in tagMap) { + failedGuids += source.guid + return@forEachIndexed + } + if (catchAllOutbound == null && routingRules.isNotEmpty()) { + failedGuids += source.guid + return@forEachIndexed + } + val routedTag = catchAllOutbound + val runtimeTag = routedTag ?: tagMap.keys.first() + OutboundProbeProfilePlan( + guid = source.guid, + outboundTags = listOf(tagMap.getValue(runtimeTag)), + ) + } + + if (profile == null || profile.outboundTags.isEmpty()) { + failedGuids += source.guid + } else { + if (originalBalancer?.obj("strategy")?.string("type") + ?.equals("leastLoad", ignoreCase = true) == true + ) { + root.obj("burstObservatory")?.obj("pingConfig")?.let { pingConfig -> + batchSamples = maxOf(batchSamples, pingConfig.positiveInt("sampling") ?: 1) + batchTimeout = longerDuration( + batchTimeout, + pingConfig.string("timeout") ?: DEFAULT_PROBE_TIMEOUT, + ) + val method = pingConfig.string("httpMethod") + ?.uppercase() + ?.takeIf { it == "GET" || it == "HEAD" } + ?: "HEAD" + leastLoadHttpMethod = when { + leastLoadHttpMethod == null -> method + leastLoadHttpMethod == "GET" || method == "GET" -> "GET" + else -> "HEAD" + } + } + } + outboundObjects.forEach { mergedOutbounds.add(it) } + localBalancers.forEach { mergedBalancers.add(it) } + profiles += profile + } + } + + val root = JsonObject().apply { + add("log", JsonObject().apply { addProperty("loglevel", "warning") }) + add("outbounds", mergedOutbounds) + add("routing", JsonObject().apply { + addProperty("domainStrategy", "AsIs") + add("rules", JsonArray()) + if (mergedBalancers.size() > 0) add("balancers", mergedBalancers) + }) + add("burstObservatory", JsonObject().apply { + add("subjectSelector", JsonArray()) + add("pingConfig", JsonObject().apply { + addProperty("destination", destination) + addProperty("httpMethod", leastLoadHttpMethod ?: DEFAULT_HTTP_METHOD) + addProperty("interval", "1h") + addProperty("sampling", batchSamples) + addProperty("timeout", batchTimeout) + }) + }) + } + + return OutboundProbePlan( + content = JsonUtil.toJsonPretty(root).orEmpty(), + profiles = profiles, + failedGuids = failedGuids, + samples = batchSamples, + ) + } + + private fun buildPolicyProfile( + guid: String, + namespace: String, + sourceBalancer: JsonObject, + tagMap: Map, + mergedBalancers: JsonArray, + ): OutboundProbeProfilePlan? { + val strategy = sourceBalancer.obj("strategy") ?: return null + val strategyType = strategy.string("type") + if (strategyType?.equals("leastPing", ignoreCase = true) != true && + strategyType?.equals("leastLoad", ignoreCase = true) != true + ) return null + if ((strategy.obj("settings")?.array("costs")?.size() ?: 0) > 0) { + // Cost matchers refer to original outbound tags. Silently carrying + // them into a namespaced batch would change leastLoad semantics. + return null + } + + val selectors = sourceBalancer.array("selector") + ?.mapNotNull { it.takeIf { selector -> selector.isJsonPrimitive }?.asString } + .orEmpty() + val outboundTags = tagMap.entries + .filter { (runtimeTag, _) -> selectors.any(runtimeTag::startsWith) } + .map { (_, probeTag) -> probeTag } + if (outboundTags.isEmpty()) return null + + val balancer = sourceBalancer.deepCopy() + val probeBalancerTag = "$namespace${sourceBalancer.string("tag").orEmpty()}" + balancer.addProperty("tag", probeBalancerTag) + balancer.add("selector", JsonArray().apply { + selectors.forEach { add("$namespace$it") } + }) + sourceBalancer.string("fallbackTag")?.let { fallback -> + val mappedFallback = tagMap[fallback] ?: return null + balancer.addProperty("fallbackTag", mappedFallback) + } + mergedBalancers.add(balancer) + return OutboundProbeProfilePlan(guid, outboundTags, probeBalancerTag) + } + + private fun remapOutboundReferences(outbound: JsonObject, tagMap: Map): Boolean { + outbound.obj("streamSettings") + ?.obj("sockopt") + ?.let { sockopt -> + sockopt.string("dialerProxy")?.let { old -> + if (old.isNotBlank()) { + val mapped = tagMap[old] ?: return false + sockopt.addProperty("dialerProxy", mapped) + } + } + } + outbound.obj("proxySettings")?.let { proxySettings -> + proxySettings.string("tag")?.let { old -> + if (old.isNotBlank()) { + val mapped = tagMap[old] ?: return false + proxySettings.addProperty("tag", mapped) + } + } + } + return true + } + + private fun JsonObject.obj(name: String): JsonObject? = + get(name)?.takeIf { it.isJsonObject }?.asJsonObject + + private fun JsonObject.array(name: String): JsonArray? = + get(name)?.takeIf { it.isJsonArray }?.asJsonArray + + private fun JsonObject.string(name: String): String? = + get(name)?.takeIf { it.isJsonPrimitive && it.asJsonPrimitive.isString }?.asString + + private fun JsonObject.positiveInt(name: String): Int? = + get(name)?.takeIf { it.isJsonPrimitive } + ?.runCatching { asInt } + ?.getOrNull() + ?.takeIf { it > 0 } + + private fun longerDuration(first: String, second: String): String { + val firstMillis = durationMillis(first) ?: return second + val secondMillis = durationMillis(second) ?: return first + return if (secondMillis > firstMillis) second else first + } + + private fun durationMillis(value: String): Long? { + val match = DURATION_PATTERN.matchEntire(value.trim()) ?: return null + val amount = match.groupValues[1].toLongOrNull() ?: return null + val multiplier = when (match.groupValues[2]) { + "ms" -> 1L + "s" -> 1_000L + "m" -> 60_000L + "h" -> 3_600_000L + else -> return null + } + return if (amount <= Long.MAX_VALUE / multiplier) amount * multiplier else Long.MAX_VALUE + } + + private fun JsonObject.isCatchAllRule(): Boolean { + val constrainedFields = listOf( + "domain", "ip", "port", "sourcePort", "source", "user", "inboundTag", + "protocol", "attrs", "process", + ) + if (constrainedFields.any(::has)) return false + val network = string("network") + return network == null || network == "tcp,udp" || network == "tcp, udp" + } + + private val DURATION_PATTERN = Regex("""([1-9]\d*)(ms|s|m|h)""") + private const val DEFAULT_HTTP_METHOD = "GET" + private const val DEFAULT_PROBE_TIMEOUT = "5s" +} diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/dto/OutboundProbePlan.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/dto/OutboundProbePlan.kt new file mode 100644 index 0000000000..4b2249381d --- /dev/null +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/dto/OutboundProbePlan.kt @@ -0,0 +1,14 @@ +package com.v2ray.ang.dto + +data class OutboundProbeProfilePlan( + val guid: String, + val outboundTags: List, + val balancerTag: String? = null, +) + +data class OutboundProbePlan( + val content: String, + val profiles: List, + val failedGuids: List, + val samples: Int, +) diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/handler/SettingsManager.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/handler/SettingsManager.kt index f4cd34a68f..98f24dc360 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/handler/SettingsManager.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/handler/SettingsManager.kt @@ -379,7 +379,7 @@ object SettingsManager { /** * Get real ping concurrency. - * @return The number of concurrent real-ping tests (clamped to 1..64). + * @return The number of concurrent real-ping configuration groups (clamped to 1..128). */ fun getRealPingConcurrency(): Int { val value = MmkvManager.decodeSettingsString(AppConfig.PREF_REAL_PING_CONCURRENCY)?.toIntOrNull() ?: 16 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 1ef04d8adc..7fdf27559e 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 @@ -4,7 +4,10 @@ import android.app.PendingIntent import android.app.Service import android.content.Context import android.content.Intent +import android.os.Handler import android.os.IBinder +import android.os.Looper +import android.os.Process import androidx.core.app.NotificationCompat import com.v2ray.ang.AppConfig import com.v2ray.ang.R @@ -19,16 +22,20 @@ import com.v2ray.ang.handler.MmkvManager import com.v2ray.ang.helper.MessageHelper import com.v2ray.ang.helper.NotificationHelper import com.v2ray.ang.util.LogUtil -import java.util.Collections +import java.util.concurrent.atomic.AtomicBoolean class CoreTestService : Service() { + @Volatile + private var activeWorker: RealPingWorkerService? = null + @Volatile + private var replacementRequested = false + private var batchStarted = false + private val batchFinished = AtomicBoolean(false) override fun attachBaseContext(newBase: Context?) { super.attachBaseContext(newBase?.let(AppLocaleManager::localizedContext)) } - // manage active batch workers so each batch is independent and cancellable - private val activeWorkers = Collections.synchronizedList(mutableListOf()) private val cancelAction by lazy { val intent = Intent(this, CoreTestService::class.java).putExtra( "content", @@ -47,100 +54,105 @@ class CoreTestService : Service() { ).build() } - /** - * Initializes the V2Ray environment. - */ override fun onCreate() { super.onCreate() CoreNativeManager.initCoreEnv(this) } - /** - * Binds the service. - * @param intent The intent. - * @return The binder. - */ - override fun onBind(intent: Intent?): IBinder? { - return null - } + override fun onBind(intent: Intent?): IBinder? = null - /** - * Cleans up resources when the service is destroyed. - */ override fun onDestroy() { - LogUtil.i(AppConfig.TAG, "CoreTestService is being destroyed, cancelling ${activeWorkers.size} active workers") - // cancel any active workers - val snapshot = ArrayList(activeWorkers) - snapshot.forEach { it.cancel() } - activeWorkers.clear() + LogUtil.i(AppConfig.TAG, "CoreTestService is being destroyed") + activeWorker?.cancel() + activeWorker = null + if (!replacementRequested && batchFinished.compareAndSet(false, true)) { + MessageHelper.sendMsg2UI(this, AppConfig.MSG_MEASURE_CONFIG_FINISH, "-1") + } NotificationHelper.stopForeground(this) super.onDestroy() + // A new process for every batch prevents Xray's process-wide state from + // leaking into a later probe or overlapping the long-running VPN core. + Handler(Looper.getMainLooper()).post { Process.killProcess(Process.myPid()) } } - /** - * Handles the start command for the service. - * @param intent The intent. - * @param flags The flags. - * @param startId The start ID. - * @return The start mode. - */ override fun onStartCommand(intent: Intent?, flags: Int, startId: Int): Int { NotificationHelper.startForeground( this, NotificationChannelType.CORE_TEST, getString(R.string.app_name), getString(R.string.title_real_ping_all_server), - cancelAction + cancelAction, ) val message = intent?.serializable("content") if (message == null) { stopSelf(startId) return START_NOT_STICKY } - - when (message.key) { + return when (message.key) { AppConfig.MSG_MEASURE_CONFIG_START -> handleMeasureStart(message, startId) - AppConfig.MSG_MEASURE_CONFIG_CANCEL -> handleMeasureCancel() + AppConfig.MSG_MEASURE_CONFIG_CANCEL -> handleMeasureCancel(startId) else -> { - NotificationHelper.stopForeground(this); stopSelf(startId) + NotificationHelper.stopForeground(this) + stopSelf(startId) + START_NOT_STICKY } } - return START_NOT_STICKY } - private fun handleMeasureStart(message: TestServiceMessage, startId: Int) { - LogUtil.i(AppConfig.TAG, "CoreTestService starting worker subscription ${message.subscriptionId}") + private fun handleMeasureStart(message: TestServiceMessage, startId: Int): Int { + if (batchStarted) { + replacementRequested = true + LogUtil.i(AppConfig.TAG, "CoreTestService handing the next batch to a fresh process") + Handler(Looper.getMainLooper()).post { Process.killProcess(Process.myPid()) } + return START_REDELIVER_INTENT + } + batchStarted = true - val guidsList = when { + val guids = when { message.serverGuids.isNotEmpty() -> message.serverGuids message.subscriptionId.isNotEmpty() -> MmkvManager.decodeServerList(message.subscriptionId) else -> MmkvManager.decodeAllServerList() } - - if (guidsList.isNotEmpty()) { - lateinit var worker: RealPingWorkerService - worker = RealPingWorkerService( - context = this, - guids = guidsList, - onlyTcp = message.onlyTcp, - onEvent = { event -> handleWorkerEvent(event, message) { activeWorkers.remove(worker) } } - ) - activeWorkers.add(worker) - worker.start() - } else { + if (guids.isEmpty()) { + batchFinished.set(true) + MessageHelper.sendMsg2UI(this, AppConfig.MSG_MEASURE_CONFIG_FINISH, "0") NotificationHelper.stopForeground(this) stopSelf(startId) + return START_NOT_STICKY } + + LogUtil.i(AppConfig.TAG, "CoreTestService starting a ${guids.size}-profile batch") + activeWorker = RealPingWorkerService( + context = this, + guids = guids, + onlyTcp = message.onlyTcp, + onEvent = { event -> handleWorkerEvent(event, message) }, + ).also { it.start() } + return START_NOT_STICKY } - private fun handleWorkerEvent(event: RealPingEvent, message: TestServiceMessage, onWorkerDone: () -> Unit) { + private fun handleMeasureCancel(startId: Int): Int { + LogUtil.i(AppConfig.TAG, "CoreTestService cancelling the active batch") + replacementRequested = false + activeWorker?.cancel() + if (batchFinished.compareAndSet(false, true)) { + MessageHelper.sendMsg2UI(this, AppConfig.MSG_MEASURE_CONFIG_FINISH, "-1") + } + activeWorker = null + NotificationHelper.stopForeground(this) + stopSelf(startId) + return START_NOT_STICKY + } + + private fun handleWorkerEvent(event: RealPingEvent, message: TestServiceMessage) { + if (replacementRequested) return when (event) { is RealPingEvent.Progress -> { NotificationHelper.updateNotification( channelType = NotificationChannelType.CORE_TEST, context = this, title = getString(R.string.app_name), - content = getString(R.string.connection_running_task_left, event.text) + content = getString(R.string.connection_running_task_left, event.text), ) MessageHelper.sendMsg2UI(this, AppConfig.MSG_MEASURE_CONFIG_NOTIFY, event.text) } @@ -151,7 +163,7 @@ class CoreTestService : Service() { } is RealPingEvent.Finish -> { - if(message.subscriptionId.isNotEmpty()){ + if (message.subscriptionId.isNotEmpty()) { if (MmkvManager.decodeSettingsBool(AppConfig.PREF_AUTO_REMOVE_INVALID_AFTER_TEST, false)) { AngConfigManager.removeInvalidServer(message.subscriptionId) } @@ -160,24 +172,12 @@ class CoreTestService : Service() { AngConfigManager.sortByTestResultsForSub(message.subscriptionId) } } - + if (!batchFinished.compareAndSet(false, true)) return MessageHelper.sendMsg2UI(this, AppConfig.MSG_MEASURE_CONFIG_FINISH, event.status) - onWorkerDone() - if (activeWorkers.isEmpty()) { - NotificationHelper.stopForeground(this) - stopSelf() - } + activeWorker = null + NotificationHelper.stopForeground(this) + stopSelf() } } } - - private fun handleMeasureCancel() { - MessageHelper.sendMsg2UI(this, AppConfig.MSG_MEASURE_CONFIG_FINISH, "0") - LogUtil.i(AppConfig.TAG, "CoreTestService received cancel message, cancelling ${activeWorkers.size} active workers") - val snapshot = ArrayList(activeWorkers) - snapshot.forEach { it.cancel() } - activeWorkers.clear() - NotificationHelper.stopForeground(this) - stopSelf() - } } diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt index d37346f1ab..0823bf907b 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt @@ -1,6 +1,7 @@ package com.v2ray.ang.service import android.content.Context +import com.v2ray.ang.AppConfig import com.v2ray.ang.core.CoreConfigManager import com.v2ray.ang.core.CoreNativeManager import com.v2ray.ang.dto.RealPingEvent @@ -10,127 +11,143 @@ import com.v2ray.ang.extension.isNotNullEmpty import com.v2ray.ang.handler.MmkvManager import com.v2ray.ang.handler.SettingsManager import com.v2ray.ang.handler.SpeedtestManager +import com.v2ray.ang.util.JsonUtil +import com.v2ray.ang.util.LogUtil import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineName import kotlinx.coroutines.CoroutineScope -import kotlinx.coroutines.SupervisorJob -import kotlinx.coroutines.asCoroutineDispatcher -import kotlinx.coroutines.isActive +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.Job import kotlinx.coroutines.joinAll import kotlinx.coroutines.launch -import java.util.concurrent.Executors -import java.util.concurrent.atomic.AtomicInteger +import libv2ray.OutboundProbeHandler +import java.util.concurrent.atomic.AtomicBoolean -/** - * Worker that runs a batch of real-ping tests independently. - * Each batch owns its own CoroutineScope/dispatcher and can be cancelled separately. - */ +/** Runs one progressively reported delay-test batch through one native core. */ class RealPingWorkerService( private val context: Context, private val guids: List, private val onlyTcp: Boolean = false, - private val onEvent: (RealPingEvent) -> Unit = {} + private val onEvent: (RealPingEvent) -> Unit = {}, ) { - private val job = SupervisorJob() - private val concurrency = SettingsManager.getRealPingConcurrency() - private val dispatcher = Executors.newFixedThreadPool(if (onlyTcp) concurrency * 2 else concurrency).asCoroutineDispatcher() - private val scope = CoroutineScope(job + dispatcher + CoroutineName("RealPingBatchWorker")) - - private val runningCount = AtomicInteger(0) - private val totalCount = AtomicInteger(0) + private val job = Job() + private val scope = CoroutineScope(job + Dispatchers.IO + CoroutineName("OutboundProbeBatch")) + private val controller = CoreNativeManager.newOutboundProbeController() + private val finished = AtomicBoolean(false) + private val emittedDelays = mutableMapOf() + private val completedGuids = mutableSetOf() + private var totalProfiles = 0 fun start() { - val jobs = guids.map { guid -> - totalCount.incrementAndGet() - scope.launch { - runningCount.incrementAndGet() - try { - val result = if (onlyTcp) startTcping(guid) else startRealPing(guid) - if (scope.isActive) { - onEvent(RealPingEvent.Result(guid, result)) - } - } catch (_: Throwable) { - // ignore - } finally { - val count = totalCount.decrementAndGet() - val left = runningCount.decrementAndGet() - if (scope.isActive) { - onEvent(RealPingEvent.Progress("$left / $count")) - } + if (onlyTcp) { + startTcpBatch() + return + } + scope.launch { + try { + val plan = CoreConfigManager.getV2rayConfig4BatchSpeedtest(context, guids) + totalProfiles = (plan.profiles.map { it.guid } + plan.failedGuids).distinct().size + plan.failedGuids.forEach { emitResult(it, -1L, completed = true) } + if (plan.profiles.isNotEmpty()) { + val concurrency = SettingsManager.getRealPingConcurrency() + LogUtil.i( + AppConfig.TAG, + "Starting ${plan.profiles.size} real-delay profiles with concurrency $concurrency", + ) + controller.probe( + plan.content, + JsonUtil.toJson(plan.profiles), + concurrency, + plan.samples, + object : OutboundProbeHandler { + override fun onOutboundProbeResult( + groupID: String?, + delay: Long, + alive: Boolean, + completed: Boolean, + ): Long { + groupID?.let { + emitResult(it, if (alive) delay else -1L, completed) + } + return 0 + } + }, + ) + } + completeMissing(plan.profiles.map { it.guid } + plan.failedGuids) + finish("0") + } catch (_: CancellationException) { + finish("-1") + } catch (error: Throwable) { + if (!finished.get()) { + LogUtil.e(AppConfig.TAG, "Outbound probe batch failed", error) + finish("-1") } } } + } + private fun startTcpBatch() { + totalProfiles = guids.size + val jobs = guids.map { guid -> + scope.launch(Dispatchers.IO.limitedParallelism(SettingsManager.getRealPingConcurrency() * 2)) { + emitResult(guid, startTcping(guid), completed = true) + } + } scope.launch { try { - joinAll(*jobs.toTypedArray()) - if (isActive) { - onEvent(RealPingEvent.Finish("0")) - } + jobs.joinAll() + finish("0") } catch (_: CancellationException) { - // If cancelled, don't send finish event to avoid confusion - } finally { - close() + finish("-1") } } } fun cancel() { + controller.cancel() job.cancel() + finish("-1") } - private fun close() { - try { - dispatcher.close() - } catch (_: Throwable) { - // ignore + @Synchronized + private fun completeMissing(allGuids: List) { + allGuids.distinct().forEach { guid -> + if (guid !in completedGuids) emitResult(guid, emittedDelays[guid] ?: -1L, completed = true) } } - private fun startRealPing(guid: String): Long { - val retFailure = -1L - - val config = MmkvManager.decodeServerConfig(guid) ?: return retFailure - if (!config.configType.isComplexType() - && config.configType != EConfigType.HYSTERIA2 - && config.configType != EConfigType.WIREGUARD - && config.alpn?.startsWith("h3") != true - && config.server.isNotNullEmpty() - && config.serverPort?.toIntOrNull() != null - ) { - val url = config.server.orEmpty() - val port = config.serverPort.orEmpty().toInt() - val tcpTime = SpeedtestManager.socketConnectTime(url, port, 1000) - if (tcpTime <= -1L) { - return retFailure - } + @Synchronized + private fun emitResult(guid: String, delay: Long, completed: Boolean) { + if (finished.get()) return + if (emittedDelays[guid] != delay) { + emittedDelays[guid] = delay + onEvent(RealPingEvent.Result(guid, delay)) } + if (completed && completedGuids.add(guid)) { + val remaining = (totalProfiles - completedGuids.size).coerceAtLeast(0) + onEvent(RealPingEvent.Progress("$remaining / $totalProfiles")) + } + } - val configResult = CoreConfigManager.getV2rayConfig4Speedtest(context, guid) - if (!configResult.status) { - return retFailure + @Synchronized + private fun finish(status: String) { + if (finished.compareAndSet(false, true)) { + onEvent(RealPingEvent.Finish(status)) } - return CoreNativeManager.measureOutboundDelay(configResult.content, SettingsManager.getDelayTestUrl()) } private fun startTcping(guid: String): Long { - val retFailure = -1L - - val config = MmkvManager.decodeServerConfig(guid) ?: return retFailure - if (!config.configType.isComplexType() - && config.configType != EConfigType.HYSTERIA2 - && config.configType != EConfigType.WIREGUARD - && config.alpn?.startsWith("h3") != true - && config.server.isNotNullEmpty() - && config.serverPort?.toIntOrNull() != null + val config = MmkvManager.decodeServerConfig(guid) ?: return -1L + if (!config.configType.isComplexType() && + config.configType != EConfigType.HYSTERIA2 && + config.configType != EConfigType.WIREGUARD && + config.alpn?.startsWith("h3") != true && + config.server.isNotNullEmpty() && + config.serverPort?.toIntOrNull() != null ) { - val url = config.server.orEmpty() - val port = config.serverPort.orEmpty().toInt() - val tcpTime = SpeedtestManager.socketConnectTime(url, port, 1000) - - return tcpTime + return SpeedtestManager.socketConnectTime(config.server.orEmpty(), config.serverPort.orEmpty().toInt(), 1000) } - - return retFailure + return -1L } } diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainBottomBar.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainBottomBar.kt index ce88845557..858e4f6287 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainBottomBar.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainBottomBar.kt @@ -36,6 +36,7 @@ import com.v2ray.ang.ui.compose.colorFabInactiveLight fun MainBottomBar( displayText: String, isRunning: Boolean, + isTesting: Boolean, isDarkTheme: Boolean, onAction: (MainAction) -> Unit ) { @@ -47,7 +48,7 @@ fun MainBottomBar( .fillMaxWidth() .windowInsetsPadding(WindowInsets.navigationBars) .height(64.dp) - .clickable(onClick = { onAction(MainAction.TestCurrentServer) }), + .clickable { onAction(if (isTesting) MainAction.CancelTesting else MainAction.TestCurrentServer) }, color = MaterialTheme.colorScheme.surface, tonalElevation = 0.dp ) { diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainScreen.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainScreen.kt index 35b2cbe8bb..37138f2fa3 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainScreen.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainScreen.kt @@ -236,6 +236,7 @@ fun MainScreen( MainBottomBar( displayText = displayText, isRunning = isRunning, + isTesting = uiState.isTesting, isDarkTheme = isDarkTheme, onAction = onAction ) 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 c1114af800..03b25fca9c 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 @@ -129,6 +129,7 @@ class MainViewModel( } is MainServiceEvent.MeasureConfigNotify -> { + if (!uiState.value.isTesting) return _uiState.update { it.copy( statusText = dataSource.getString( @@ -683,7 +684,7 @@ class MainViewModel( _uiState.update { it.copy( isTesting = true, - statusText = dataSource.getString(R.string.connection_test_testing) + statusText = dataSource.getString(R.string.connection_test_testing_tap_to_stop) ) } viewModelScope.launch(ioDispatcher) { From 1859071185399d43ece9dcc119ecdd7fac7f764e Mon Sep 17 00:00:00 2001 From: Eliot the Cougar Date: Sun, 2 Aug 2026 00:14:45 +0300 Subject: [PATCH 02/10] Use concise probe names and enforce one concurrency limit --- V2rayNG/app/src/main/AndroidManifest.xml | 2 +- .../com/v2ray/ang/core/CoreConfigManager.kt | 10 +++++----- .../com/v2ray/ang/core/CoreNativeManager.kt | 6 +++--- ...ConfigBuilder.kt => ProbeConfigBuilder.kt} | 20 +++++++++---------- .../{OutboundProbePlan.kt => ProbePlan.kt} | 6 +++--- .../com/v2ray/ang/handler/SettingsManager.kt | 2 +- .../ang/service/RealPingWorkerService.kt | 18 +++++++++-------- 7 files changed, 33 insertions(+), 31 deletions(-) rename V2rayNG/app/src/main/java/com/v2ray/ang/core/{OutboundProbeConfigBuilder.kt => ProbeConfigBuilder.kt} (96%) rename V2rayNG/app/src/main/java/com/v2ray/ang/dto/{OutboundProbePlan.kt => ProbePlan.kt} (64%) diff --git a/V2rayNG/app/src/main/AndroidManifest.xml b/V2rayNG/app/src/main/AndroidManifest.xml index bfc145632d..2fbb8b45ec 100644 --- a/V2rayNG/app/src/main/AndroidManifest.xml +++ b/V2rayNG/app/src/main/AndroidManifest.xml @@ -256,7 +256,7 @@ android:name=".service.CoreTestService" android:exported="false" android:foregroundServiceType="specialUse" - android:process=":OutboundProbe"> + android:process=":Probe"> diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt index 1da3ad433e..2e3a7b9fbf 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt @@ -7,7 +7,7 @@ import com.google.gson.JsonObject import com.v2ray.ang.AppConfig import com.v2ray.ang.dto.ConfigResult import com.v2ray.ang.dto.CoreConfigContext -import com.v2ray.ang.dto.OutboundProbePlan +import com.v2ray.ang.dto.ProbePlan import com.v2ray.ang.dto.V2rayConfig import com.v2ray.ang.dto.entities.ProfileItem import com.v2ray.ang.dto.entities.RulesetItem @@ -85,18 +85,18 @@ object CoreConfigManager { } /** Builds one isolated Xray configuration for a complete UI delay-test batch. */ - fun getV2rayConfig4BatchSpeedtest(context: Context, guids: List): OutboundProbePlan { - val sources = mutableListOf() + fun getV2rayConfig4BatchSpeedtest(context: Context, guids: List): ProbePlan { + val sources = mutableListOf() val failedGuids = mutableListOf() guids.distinct().forEach { guid -> val result = getV2rayConfig4Speedtest(context, guid) if (result.status && result.content.isNotBlank()) { - sources += OutboundProbeConfigBuilder.Source(guid, result.content) + sources += ProbeConfigBuilder.Source(guid, result.content) } else { failedGuids += guid } } - val plan = OutboundProbeConfigBuilder.build( + val plan = ProbeConfigBuilder.build( sources = sources, destination = SettingsManager.getDelayTestUrl(), ) diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreNativeManager.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreNativeManager.kt index ff31ee090b..4b4f7bbecf 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreNativeManager.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreNativeManager.kt @@ -8,7 +8,7 @@ import go.Seq import libv2ray.CoreCallbackHandler import libv2ray.CoreController import libv2ray.Libv2ray -import libv2ray.OutboundProbeController +import libv2ray.ProbeController import java.util.concurrent.atomic.AtomicBoolean /** @@ -68,8 +68,8 @@ object CoreNativeManager { } } - fun newOutboundProbeController(): OutboundProbeController = - Libv2ray.newOutboundProbeController() + fun newProbeController(): ProbeController = + Libv2ray.newProbeController() /** * Measure outbound connection delay. diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/core/OutboundProbeConfigBuilder.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt similarity index 96% rename from V2rayNG/app/src/main/java/com/v2ray/ang/core/OutboundProbeConfigBuilder.kt rename to V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt index 91780b25cb..87c6b490a4 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/core/OutboundProbeConfigBuilder.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt @@ -3,8 +3,8 @@ package com.v2ray.ang.core import com.google.gson.JsonArray import com.google.gson.JsonObject -import com.v2ray.ang.dto.OutboundProbePlan -import com.v2ray.ang.dto.OutboundProbeProfilePlan +import com.v2ray.ang.dto.ProbePlan +import com.v2ray.ang.dto.ProbeProfile import com.v2ray.ang.util.JsonUtil /** @@ -12,11 +12,11 @@ import com.v2ray.ang.util.JsonUtil * * Every source gets a private tag namespace. Only outbound definitions and the * primary policy-group balancer are retained. AndroidLib drives the unchanged - * upstream BurstObservatory.Check API with profile-level concurrency and + * upstream BurstObservatory.Check API with probe-level concurrency and * progressive group results, so unrelated routing and inbound state must * not affect another profile in the same batch. */ -object OutboundProbeConfigBuilder { +object ProbeConfigBuilder { data class Source(val guid: String, val content: String) fun build( @@ -24,13 +24,13 @@ object OutboundProbeConfigBuilder { destination: String, timeout: String = DEFAULT_PROBE_TIMEOUT, samples: Int = 1, - ): OutboundProbePlan { + ): ProbePlan { require(destination.isNotBlank()) { "probe destination is empty" } require(samples > 0) { "probe sample count must be positive" } val mergedOutbounds = JsonArray() val mergedBalancers = JsonArray() - val profiles = mutableListOf() + val profiles = mutableListOf() val failedGuids = mutableListOf() var batchSamples = samples var batchTimeout = timeout @@ -116,7 +116,7 @@ object OutboundProbeConfigBuilder { } val routedTag = catchAllOutbound val runtimeTag = routedTag ?: tagMap.keys.first() - OutboundProbeProfilePlan( + ProbeProfile( guid = source.guid, outboundTags = listOf(tagMap.getValue(runtimeTag)), ) @@ -171,7 +171,7 @@ object OutboundProbeConfigBuilder { }) } - return OutboundProbePlan( + return ProbePlan( content = JsonUtil.toJsonPretty(root).orEmpty(), profiles = profiles, failedGuids = failedGuids, @@ -185,7 +185,7 @@ object OutboundProbeConfigBuilder { sourceBalancer: JsonObject, tagMap: Map, mergedBalancers: JsonArray, - ): OutboundProbeProfilePlan? { + ): ProbeProfile? { val strategy = sourceBalancer.obj("strategy") ?: return null val strategyType = strategy.string("type") if (strategyType?.equals("leastPing", ignoreCase = true) != true && @@ -216,7 +216,7 @@ object OutboundProbeConfigBuilder { balancer.addProperty("fallbackTag", mappedFallback) } mergedBalancers.add(balancer) - return OutboundProbeProfilePlan(guid, outboundTags, probeBalancerTag) + return ProbeProfile(guid, outboundTags, probeBalancerTag) } private fun remapOutboundReferences(outbound: JsonObject, tagMap: Map): Boolean { diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/dto/OutboundProbePlan.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/dto/ProbePlan.kt similarity index 64% rename from V2rayNG/app/src/main/java/com/v2ray/ang/dto/OutboundProbePlan.kt rename to V2rayNG/app/src/main/java/com/v2ray/ang/dto/ProbePlan.kt index 4b2249381d..b212f04a53 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/dto/OutboundProbePlan.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/dto/ProbePlan.kt @@ -1,14 +1,14 @@ package com.v2ray.ang.dto -data class OutboundProbeProfilePlan( +data class ProbeProfile( val guid: String, val outboundTags: List, val balancerTag: String? = null, ) -data class OutboundProbePlan( +data class ProbePlan( val content: String, - val profiles: List, + val profiles: List, val failedGuids: List, val samples: Int, ) diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/handler/SettingsManager.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/handler/SettingsManager.kt index 98f24dc360..42fe31f147 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/handler/SettingsManager.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/handler/SettingsManager.kt @@ -379,7 +379,7 @@ object SettingsManager { /** * Get real ping concurrency. - * @return The number of concurrent real-ping configuration groups (clamped to 1..128). + * @return The maximum number of simultaneous real-delay probes (clamped to 1..128). */ fun getRealPingConcurrency(): Int { val value = MmkvManager.decodeSettingsString(AppConfig.PREF_REAL_PING_CONCURRENCY)?.toIntOrNull() ?: 16 diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt index 0823bf907b..30414a0e73 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt @@ -20,7 +20,7 @@ import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job import kotlinx.coroutines.joinAll import kotlinx.coroutines.launch -import libv2ray.OutboundProbeHandler +import libv2ray.ProbeHandler import java.util.concurrent.atomic.AtomicBoolean /** Runs one progressively reported delay-test batch through one native core. */ @@ -31,8 +31,8 @@ class RealPingWorkerService( private val onEvent: (RealPingEvent) -> Unit = {}, ) { private val job = Job() - private val scope = CoroutineScope(job + Dispatchers.IO + CoroutineName("OutboundProbeBatch")) - private val controller = CoreNativeManager.newOutboundProbeController() + private val scope = CoroutineScope(job + Dispatchers.IO + CoroutineName("ProbeBatch")) + private val controller = CoreNativeManager.newProbeController() private val finished = AtomicBoolean(false) private val emittedDelays = mutableMapOf() private val completedGuids = mutableSetOf() @@ -50,17 +50,18 @@ class RealPingWorkerService( plan.failedGuids.forEach { emitResult(it, -1L, completed = true) } if (plan.profiles.isNotEmpty()) { val concurrency = SettingsManager.getRealPingConcurrency() + val probeCount = plan.profiles.sumOf { it.outboundTags.size } LogUtil.i( AppConfig.TAG, - "Starting ${plan.profiles.size} real-delay profiles with concurrency $concurrency", + "Starting $probeCount real-delay probes for ${plan.profiles.size} profiles with limit $concurrency", ) controller.probe( plan.content, JsonUtil.toJson(plan.profiles), concurrency, plan.samples, - object : OutboundProbeHandler { - override fun onOutboundProbeResult( + object : ProbeHandler { + override fun onProbeResult( groupID: String?, delay: Long, alive: Boolean, @@ -80,7 +81,7 @@ class RealPingWorkerService( finish("-1") } catch (error: Throwable) { if (!finished.get()) { - LogUtil.e(AppConfig.TAG, "Outbound probe batch failed", error) + LogUtil.e(AppConfig.TAG, "Probe batch failed", error) finish("-1") } } @@ -89,8 +90,9 @@ class RealPingWorkerService( private fun startTcpBatch() { totalProfiles = guids.size + val dispatcher = Dispatchers.IO.limitedParallelism(SettingsManager.getRealPingConcurrency()) val jobs = guids.map { guid -> - scope.launch(Dispatchers.IO.limitedParallelism(SettingsManager.getRealPingConcurrency() * 2)) { + scope.launch(dispatcher) { emitResult(guid, startTcping(guid), completed = true) } } From 5fd497450c0cf611457da3ec009424583abbec54 Mon Sep 17 00:00:00 2001 From: Eliot the Cougar Date: Sun, 2 Aug 2026 00:34:51 +0300 Subject: [PATCH 03/10] Use one sample for each Observatory target --- .../java/com/v2ray/ang/core/ProbeConfigBuilder.kt | 13 +------------ .../src/main/java/com/v2ray/ang/dto/ProbePlan.kt | 1 - .../com/v2ray/ang/service/RealPingWorkerService.kt | 1 - 3 files changed, 1 insertion(+), 14 deletions(-) diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt index 87c6b490a4..180cd0c02d 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt @@ -23,16 +23,13 @@ object ProbeConfigBuilder { sources: List, destination: String, timeout: String = DEFAULT_PROBE_TIMEOUT, - samples: Int = 1, ): ProbePlan { require(destination.isNotBlank()) { "probe destination is empty" } - require(samples > 0) { "probe sample count must be positive" } val mergedOutbounds = JsonArray() val mergedBalancers = JsonArray() val profiles = mutableListOf() val failedGuids = mutableListOf() - var batchSamples = samples var batchTimeout = timeout var leastLoadHttpMethod: String? = null @@ -129,7 +126,6 @@ object ProbeConfigBuilder { ?.equals("leastLoad", ignoreCase = true) == true ) { root.obj("burstObservatory")?.obj("pingConfig")?.let { pingConfig -> - batchSamples = maxOf(batchSamples, pingConfig.positiveInt("sampling") ?: 1) batchTimeout = longerDuration( batchTimeout, pingConfig.string("timeout") ?: DEFAULT_PROBE_TIMEOUT, @@ -165,7 +161,7 @@ object ProbeConfigBuilder { addProperty("destination", destination) addProperty("httpMethod", leastLoadHttpMethod ?: DEFAULT_HTTP_METHOD) addProperty("interval", "1h") - addProperty("sampling", batchSamples) + addProperty("sampling", 1) addProperty("timeout", batchTimeout) }) }) @@ -175,7 +171,6 @@ object ProbeConfigBuilder { content = JsonUtil.toJsonPretty(root).orEmpty(), profiles = profiles, failedGuids = failedGuids, - samples = batchSamples, ) } @@ -250,12 +245,6 @@ object ProbeConfigBuilder { private fun JsonObject.string(name: String): String? = get(name)?.takeIf { it.isJsonPrimitive && it.asJsonPrimitive.isString }?.asString - private fun JsonObject.positiveInt(name: String): Int? = - get(name)?.takeIf { it.isJsonPrimitive } - ?.runCatching { asInt } - ?.getOrNull() - ?.takeIf { it > 0 } - private fun longerDuration(first: String, second: String): String { val firstMillis = durationMillis(first) ?: return second val secondMillis = durationMillis(second) ?: return first diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/dto/ProbePlan.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/dto/ProbePlan.kt index b212f04a53..e591e2bc1c 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/dto/ProbePlan.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/dto/ProbePlan.kt @@ -10,5 +10,4 @@ data class ProbePlan( val content: String, val profiles: List, val failedGuids: List, - val samples: Int, ) diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt index 30414a0e73..eb51b96add 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt @@ -59,7 +59,6 @@ class RealPingWorkerService( plan.content, JsonUtil.toJson(plan.profiles), concurrency, - plan.samples, object : ProbeHandler { override fun onProbeResult( groupID: String?, From 4f3c66b3399ef0e4fde56aec7245e9ce6ffa0fc1 Mon Sep 17 00:00:00 2001 From: Eliot the Cougar Date: Sun, 2 Aug 2026 15:41:05 +0300 Subject: [PATCH 04/10] Simplify app-side Observatory probing Build batch configs directly from v2rayNG's typed speed-test models instead of reparsing and validating arbitrary JSON states the app does not generate. Keep custom configs and non-Observatory policy strategies on the existing individual delay path. Use the worker's single completion contract, retain only cancellation and pending-result state that can occur, and isolate subscription-update probes from the live VPN daemon process. --- V2rayNG/app/src/main/AndroidManifest.xml | 2 +- .../com/v2ray/ang/core/CoreConfigManager.kt | 31 +- .../com/v2ray/ang/core/CoreNativeManager.kt | 4 - .../com/v2ray/ang/core/ProbeConfigBuilder.kt | 318 ++++-------------- .../main/java/com/v2ray/ang/dto/ProbePlan.kt | 3 +- .../com/v2ray/ang/handler/SettingsManager.kt | 2 +- .../com/v2ray/ang/service/CoreTestService.kt | 7 - .../ang/service/RealPingWorkerService.kt | 73 ++-- 8 files changed, 140 insertions(+), 300 deletions(-) diff --git a/V2rayNG/app/src/main/AndroidManifest.xml b/V2rayNG/app/src/main/AndroidManifest.xml index 2fbb8b45ec..4b397870c0 100644 --- a/V2rayNG/app/src/main/AndroidManifest.xml +++ b/V2rayNG/app/src/main/AndroidManifest.xml @@ -266,7 +266,7 @@ android:name=".service.SubscriptionUpdateService" android:exported="false" android:foregroundServiceType="specialUse" - android:process=":RunSoLibV2RayDaemon"> + android:process=":SubscriptionUpdate"> diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt index 2e3a7b9fbf..db9376fc39 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt @@ -70,10 +70,7 @@ object CoreConfigManager { if (configContext.isCustom) { return buildV2rayCustomConfig(configContext) } - val v2rayConfig = buildUnifiedConfig(configContext) - postProcessForSpeedtest(v2rayConfig) - - return toConfigResult(configContext, v2rayConfig) + return toConfigResult(configContext, buildSpeedtestConfig(configContext)) } catch (e: Exception) { LogUtil.e(AppConfig.TAG, "Failed to get V2ray config for speedtest", e) return ConfigResult( @@ -85,14 +82,22 @@ object CoreConfigManager { } /** Builds one isolated Xray configuration for a complete UI delay-test batch. */ - fun getV2rayConfig4BatchSpeedtest(context: Context, guids: List): ProbePlan { + fun getProbePlan(context: Context, guids: List): ProbePlan { val sources = mutableListOf() + val individualGuids = mutableListOf() val failedGuids = mutableListOf() guids.distinct().forEach { guid -> - val result = getV2rayConfig4Speedtest(context, guid) - if (result.status && result.content.isNotBlank()) { - sources += ProbeConfigBuilder.Source(guid, result.content) - } else { + try { + val configContext = CoreConfigContextBuilder.build(context, guid) + if (configContext == null) { + failedGuids += guid + } else if (configContext.isCustom) { + individualGuids += guid + } else { + sources += ProbeConfigBuilder.Source(guid, buildSpeedtestConfig(configContext)) + } + } catch (error: Exception) { + LogUtil.e(AppConfig.TAG, "Failed to build probe config for $guid", error) failedGuids += guid } } @@ -100,9 +105,15 @@ object CoreConfigManager { sources = sources, destination = SettingsManager.getDelayTestUrl(), ) - return plan.copy(failedGuids = (failedGuids + plan.failedGuids).distinct()) + return plan.copy( + individualGuids = (individualGuids + plan.individualGuids).distinct(), + failedGuids = failedGuids, + ) } + private fun buildSpeedtestConfig(configContext: CoreConfigContext): V2rayConfig = + buildUnifiedConfig(configContext).also(::postProcessForSpeedtest) + /** * Build configuration for custom profiles. */ diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreNativeManager.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreNativeManager.kt index 4b4f7bbecf..502870bd9f 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreNativeManager.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreNativeManager.kt @@ -8,7 +8,6 @@ import go.Seq import libv2ray.CoreCallbackHandler import libv2ray.CoreController import libv2ray.Libv2ray -import libv2ray.ProbeController import java.util.concurrent.atomic.AtomicBoolean /** @@ -68,9 +67,6 @@ object CoreNativeManager { } } - fun newProbeController(): ProbeController = - Libv2ray.newProbeController() - /** * Measure outbound connection delay. * diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt index 180cd0c02d..250167d969 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt @@ -1,280 +1,96 @@ - package com.v2ray.ang.core -import com.google.gson.JsonArray -import com.google.gson.JsonObject +import com.v2ray.ang.AppConfig import com.v2ray.ang.dto.ProbePlan import com.v2ray.ang.dto.ProbeProfile +import com.v2ray.ang.dto.V2rayConfig import com.v2ray.ang.util.JsonUtil -/** - * Combines independently generated speed-test configurations into one core. - * - * Every source gets a private tag namespace. Only outbound definitions and the - * primary policy-group balancer are retained. AndroidLib drives the unchanged - * upstream BurstObservatory.Check API with probe-level concurrency and - * progressive group results, so unrelated routing and inbound state must - * not affect another profile in the same batch. - */ +/** Combines v2rayNG-generated speed-test configurations into one probe core. */ object ProbeConfigBuilder { - data class Source(val guid: String, val content: String) - - fun build( - sources: List, - destination: String, - timeout: String = DEFAULT_PROBE_TIMEOUT, - ): ProbePlan { - require(destination.isNotBlank()) { "probe destination is empty" } + data class Source(val guid: String, val config: V2rayConfig) - val mergedOutbounds = JsonArray() - val mergedBalancers = JsonArray() + fun build(sources: List, destination: String): ProbePlan { + val outbounds = mutableListOf() + val balancers = mutableListOf() val profiles = mutableListOf() - val failedGuids = mutableListOf() - var batchTimeout = timeout - var leastLoadHttpMethod: String? = null - - sources.distinctBy { it.guid }.forEachIndexed { index, source -> - val root = JsonUtil.parseString(source.content) - val outbounds = root?.array("outbounds") - if (outbounds == null || outbounds.size() == 0) { - failedGuids += source.guid - return@forEachIndexed - } + val individualGuids = mutableListOf() + var httpMethod = DEFAULT_HTTP_METHOD + var timeout = DEFAULT_TIMEOUT + sources.forEachIndexed { index, source -> val namespace = "probe-$index-" - val outboundElements = outbounds.toList() - if (outboundElements.any { !it.isJsonObject }) { - failedGuids += source.guid - return@forEachIndexed - } - val outboundObjects = outboundElements.map { it.asJsonObject.deepCopy() } - val originalTags = outboundObjects.mapIndexed { outboundIndex, outbound -> - outbound.string("tag") ?: "outbound-$outboundIndex" - } - if (originalTags.toSet().size != originalTags.size) { - failedGuids += source.guid - return@forEachIndexed - } - - val tagMap = linkedMapOf() - outboundObjects.forEachIndexed { outboundIndex, outbound -> - val originalTag = originalTags[outboundIndex] - val probeTag = "$namespace$originalTag" - tagMap[originalTag] = probeTag - outbound.addProperty("tag", probeTag) - } - if (outboundObjects.any { !remapOutboundReferences(it, tagMap) }) { - failedGuids += source.guid - return@forEachIndexed - } - - val routing = root.obj("routing") - val routingRules = routing?.array("rules") - ?.mapNotNull { it.takeIf { rule -> rule.isJsonObject }?.asJsonObject } - .orEmpty() - val primaryBalancerTag = routingRules - .lastOrNull { it.isCatchAllRule() && it.string("balancerTag") != null } - ?.string("balancerTag") + val tagMap = source.config.outbounds.associate { it.tag to "$namespace${it.tag}" } + val primaryBalancer = source.config.routing.balancers + ?.firstOrNull { it.tag == AppConfig.TAG_BALANCER } + val strategyType = primaryBalancer?.strategy?.type - val originalBalancer = primaryBalancerTag?.let { wanted -> - routing?.array("balancers") - ?.mapNotNull { it.takeIf { balancer -> balancer.isJsonObject }?.asJsonObject } - ?.firstOrNull { it.string("tag") == wanted } - } - if ((primaryBalancerTag != null && originalBalancer == null) || - (primaryBalancerTag == null && routingRules.any { it.string("balancerTag") != null }) - ) { - // Direct batch probing cannot reproduce conditional routing. - // Accept only the catch-all policy-group form emitted by v2rayNG. - failedGuids += source.guid + if (primaryBalancer != null && strategyType !in OBSERVATORY_STRATEGIES) { + individualGuids += source.guid return@forEachIndexed } - val localBalancers = JsonArray() - val profile = if (originalBalancer != null) { - buildPolicyProfile( - source.guid, - namespace, - originalBalancer, - tagMap, - localBalancers, - ) - } else { - val catchAllOutbound = routingRules.lastOrNull { - it.isCatchAllRule() && it.string("outboundTag") != null - }?.string("outboundTag") - if (catchAllOutbound != null && catchAllOutbound !in tagMap) { - failedGuids += source.guid - return@forEachIndexed - } - if (catchAllOutbound == null && routingRules.isNotEmpty()) { - failedGuids += source.guid - return@forEachIndexed + source.config.outbounds.forEach { outbound -> + outbound.tag = tagMap.getValue(outbound.tag) + outbound.streamSettings?.sockopt?.dialerProxy?.let { dialerProxy -> + outbound.streamSettings?.sockopt?.dialerProxy = tagMap.getValue(dialerProxy) } - val routedTag = catchAllOutbound - val runtimeTag = routedTag ?: tagMap.keys.first() - ProbeProfile( - guid = source.guid, - outboundTags = listOf(tagMap.getValue(runtimeTag)), - ) + outbounds += outbound } - if (profile == null || profile.outboundTags.isEmpty()) { - failedGuids += source.guid - } else { - if (originalBalancer?.obj("strategy")?.string("type") - ?.equals("leastLoad", ignoreCase = true) == true - ) { - root.obj("burstObservatory")?.obj("pingConfig")?.let { pingConfig -> - batchTimeout = longerDuration( - batchTimeout, - pingConfig.string("timeout") ?: DEFAULT_PROBE_TIMEOUT, - ) - val method = pingConfig.string("httpMethod") - ?.uppercase() - ?.takeIf { it == "GET" || it == "HEAD" } - ?: "HEAD" - leastLoadHttpMethod = when { - leastLoadHttpMethod == null -> method - leastLoadHttpMethod == "GET" || method == "GET" -> "GET" - else -> "HEAD" - } - } - } - outboundObjects.forEach { mergedOutbounds.add(it) } - localBalancers.forEach { mergedBalancers.add(it) } - profiles += profile + if (primaryBalancer == null) { + profiles += ProbeProfile(source.guid, listOf(tagMap.getValue(AppConfig.TAG_PROXY))) + return@forEachIndexed } - } - val root = JsonObject().apply { - add("log", JsonObject().apply { addProperty("loglevel", "warning") }) - add("outbounds", mergedOutbounds) - add("routing", JsonObject().apply { - addProperty("domainStrategy", "AsIs") - add("rules", JsonArray()) - if (mergedBalancers.size() > 0) add("balancers", mergedBalancers) - }) - add("burstObservatory", JsonObject().apply { - add("subjectSelector", JsonArray()) - add("pingConfig", JsonObject().apply { - addProperty("destination", destination) - addProperty("httpMethod", leastLoadHttpMethod ?: DEFAULT_HTTP_METHOD) - addProperty("interval", "1h") - addProperty("sampling", 1) - addProperty("timeout", batchTimeout) - }) - }) + val outboundTags = tagMap + .filterKeys { tag -> primaryBalancer.selector.any(tag::startsWith) } + .values + .toList() + val probeBalancer = primaryBalancer.copy( + tag = "$namespace${primaryBalancer.tag}", + selector = primaryBalancer.selector.map { "$namespace$it" }, + fallbackTag = primaryBalancer.fallbackTag?.let(tagMap::getValue), + ) + balancers += probeBalancer + profiles += ProbeProfile(source.guid, outboundTags, probeBalancer.tag) + + if (strategyType == "leastLoad") { + val pingConfig = (source.config.burstObservatory as V2rayConfig.BurstObservatoryObject).pingConfig + httpMethod = pingConfig.httpMethod ?: DEFAULT_HTTP_METHOD + timeout = pingConfig.timeout ?: DEFAULT_TIMEOUT + } } + val routing = mutableMapOf( + "domainStrategy" to "AsIs", + "rules" to emptyList(), + ) + if (balancers.isNotEmpty()) routing["balancers"] = balancers + + val config = mapOf( + "log" to mapOf("loglevel" to "warning"), + "outbounds" to outbounds, + "routing" to routing, + "burstObservatory" to mapOf( + "subjectSelector" to emptyList(), + "pingConfig" to mapOf( + "destination" to destination, + "httpMethod" to httpMethod, + "interval" to "1h", + "sampling" to 1, + "timeout" to timeout, + ), + ), + ) return ProbePlan( - content = JsonUtil.toJsonPretty(root).orEmpty(), + content = JsonUtil.toJsonPretty(config).orEmpty(), profiles = profiles, - failedGuids = failedGuids, - ) - } - - private fun buildPolicyProfile( - guid: String, - namespace: String, - sourceBalancer: JsonObject, - tagMap: Map, - mergedBalancers: JsonArray, - ): ProbeProfile? { - val strategy = sourceBalancer.obj("strategy") ?: return null - val strategyType = strategy.string("type") - if (strategyType?.equals("leastPing", ignoreCase = true) != true && - strategyType?.equals("leastLoad", ignoreCase = true) != true - ) return null - if ((strategy.obj("settings")?.array("costs")?.size() ?: 0) > 0) { - // Cost matchers refer to original outbound tags. Silently carrying - // them into a namespaced batch would change leastLoad semantics. - return null - } - - val selectors = sourceBalancer.array("selector") - ?.mapNotNull { it.takeIf { selector -> selector.isJsonPrimitive }?.asString } - .orEmpty() - val outboundTags = tagMap.entries - .filter { (runtimeTag, _) -> selectors.any(runtimeTag::startsWith) } - .map { (_, probeTag) -> probeTag } - if (outboundTags.isEmpty()) return null - - val balancer = sourceBalancer.deepCopy() - val probeBalancerTag = "$namespace${sourceBalancer.string("tag").orEmpty()}" - balancer.addProperty("tag", probeBalancerTag) - balancer.add("selector", JsonArray().apply { - selectors.forEach { add("$namespace$it") } - }) - sourceBalancer.string("fallbackTag")?.let { fallback -> - val mappedFallback = tagMap[fallback] ?: return null - balancer.addProperty("fallbackTag", mappedFallback) - } - mergedBalancers.add(balancer) - return ProbeProfile(guid, outboundTags, probeBalancerTag) - } - - private fun remapOutboundReferences(outbound: JsonObject, tagMap: Map): Boolean { - outbound.obj("streamSettings") - ?.obj("sockopt") - ?.let { sockopt -> - sockopt.string("dialerProxy")?.let { old -> - if (old.isNotBlank()) { - val mapped = tagMap[old] ?: return false - sockopt.addProperty("dialerProxy", mapped) - } - } - } - outbound.obj("proxySettings")?.let { proxySettings -> - proxySettings.string("tag")?.let { old -> - if (old.isNotBlank()) { - val mapped = tagMap[old] ?: return false - proxySettings.addProperty("tag", mapped) - } - } - } - return true - } - - private fun JsonObject.obj(name: String): JsonObject? = - get(name)?.takeIf { it.isJsonObject }?.asJsonObject - - private fun JsonObject.array(name: String): JsonArray? = - get(name)?.takeIf { it.isJsonArray }?.asJsonArray - - private fun JsonObject.string(name: String): String? = - get(name)?.takeIf { it.isJsonPrimitive && it.asJsonPrimitive.isString }?.asString - - private fun longerDuration(first: String, second: String): String { - val firstMillis = durationMillis(first) ?: return second - val secondMillis = durationMillis(second) ?: return first - return if (secondMillis > firstMillis) second else first - } - - private fun durationMillis(value: String): Long? { - val match = DURATION_PATTERN.matchEntire(value.trim()) ?: return null - val amount = match.groupValues[1].toLongOrNull() ?: return null - val multiplier = when (match.groupValues[2]) { - "ms" -> 1L - "s" -> 1_000L - "m" -> 60_000L - "h" -> 3_600_000L - else -> return null - } - return if (amount <= Long.MAX_VALUE / multiplier) amount * multiplier else Long.MAX_VALUE - } - - private fun JsonObject.isCatchAllRule(): Boolean { - val constrainedFields = listOf( - "domain", "ip", "port", "sourcePort", "source", "user", "inboundTag", - "protocol", "attrs", "process", + individualGuids = individualGuids, ) - if (constrainedFields.any(::has)) return false - val network = string("network") - return network == null || network == "tcp,udp" || network == "tcp, udp" } - private val DURATION_PATTERN = Regex("""([1-9]\d*)(ms|s|m|h)""") + private val OBSERVATORY_STRATEGIES = setOf("leastPing", "leastLoad") private const val DEFAULT_HTTP_METHOD = "GET" - private const val DEFAULT_PROBE_TIMEOUT = "5s" + private const val DEFAULT_TIMEOUT = "5s" } diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/dto/ProbePlan.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/dto/ProbePlan.kt index e591e2bc1c..32b0a9f531 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/dto/ProbePlan.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/dto/ProbePlan.kt @@ -9,5 +9,6 @@ data class ProbeProfile( data class ProbePlan( val content: String, val profiles: List, - val failedGuids: List, + val individualGuids: List = emptyList(), + val failedGuids: List = emptyList(), ) diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/handler/SettingsManager.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/handler/SettingsManager.kt index 42fe31f147..f4cd34a68f 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/handler/SettingsManager.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/handler/SettingsManager.kt @@ -379,7 +379,7 @@ object SettingsManager { /** * Get real ping concurrency. - * @return The maximum number of simultaneous real-delay probes (clamped to 1..128). + * @return The number of concurrent real-ping tests (clamped to 1..64). */ fun getRealPingConcurrency(): Int { val value = MmkvManager.decodeSettingsString(AppConfig.PREF_REAL_PING_CONCURRENCY)?.toIntOrNull() ?: 16 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 7fdf27559e..582abd08a6 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 @@ -22,7 +22,6 @@ import com.v2ray.ang.handler.MmkvManager import com.v2ray.ang.helper.MessageHelper import com.v2ray.ang.helper.NotificationHelper import com.v2ray.ang.util.LogUtil -import java.util.concurrent.atomic.AtomicBoolean class CoreTestService : Service() { @Volatile @@ -30,7 +29,6 @@ class CoreTestService : Service() { @Volatile private var replacementRequested = false private var batchStarted = false - private val batchFinished = AtomicBoolean(false) override fun attachBaseContext(newBase: Context?) { super.attachBaseContext(newBase?.let(AppLocaleManager::localizedContext)) @@ -65,9 +63,6 @@ class CoreTestService : Service() { LogUtil.i(AppConfig.TAG, "CoreTestService is being destroyed") activeWorker?.cancel() activeWorker = null - if (!replacementRequested && batchFinished.compareAndSet(false, true)) { - MessageHelper.sendMsg2UI(this, AppConfig.MSG_MEASURE_CONFIG_FINISH, "-1") - } NotificationHelper.stopForeground(this) super.onDestroy() // A new process for every batch prevents Xray's process-wide state from @@ -114,7 +109,6 @@ class CoreTestService : Service() { else -> MmkvManager.decodeAllServerList() } if (guids.isEmpty()) { - batchFinished.set(true) MessageHelper.sendMsg2UI(this, AppConfig.MSG_MEASURE_CONFIG_FINISH, "0") NotificationHelper.stopForeground(this) stopSelf(startId) @@ -172,7 +166,6 @@ class CoreTestService : Service() { AngConfigManager.sortByTestResultsForSub(message.subscriptionId) } } - if (!batchFinished.compareAndSet(false, true)) return MessageHelper.sendMsg2UI(this, AppConfig.MSG_MEASURE_CONFIG_FINISH, event.status) activeWorker = null NotificationHelper.stopForeground(this) diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt index eb51b96add..099b2c029b 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt @@ -20,23 +20,24 @@ import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job import kotlinx.coroutines.joinAll import kotlinx.coroutines.launch +import libv2ray.Libv2ray import libv2ray.ProbeHandler -import java.util.concurrent.atomic.AtomicBoolean /** Runs one progressively reported delay-test batch through one native core. */ class RealPingWorkerService( private val context: Context, - private val guids: List, + guids: List, private val onlyTcp: Boolean = false, private val onEvent: (RealPingEvent) -> Unit = {}, ) { + private val guids = guids.distinct() private val job = Job() private val scope = CoroutineScope(job + Dispatchers.IO + CoroutineName("ProbeBatch")) - private val controller = CoreNativeManager.newProbeController() - private val finished = AtomicBoolean(false) + private val controller = Libv2ray.newProbeController() + @Volatile + private var finished = false private val emittedDelays = mutableMapOf() - private val completedGuids = mutableSetOf() - private var totalProfiles = 0 + private val pendingGuids = guids.toMutableSet() fun start() { if (onlyTcp) { @@ -45,8 +46,7 @@ class RealPingWorkerService( } scope.launch { try { - val plan = CoreConfigManager.getV2rayConfig4BatchSpeedtest(context, guids) - totalProfiles = (plan.profiles.map { it.guid } + plan.failedGuids).distinct().size + val plan = CoreConfigManager.getProbePlan(context, guids) plan.failedGuids.forEach { emitResult(it, -1L, completed = true) } if (plan.profiles.isNotEmpty()) { val concurrency = SettingsManager.getRealPingConcurrency() @@ -66,29 +66,27 @@ class RealPingWorkerService( alive: Boolean, completed: Boolean, ): Long { - groupID?.let { - emitResult(it, if (alive) delay else -1L, completed) - } + emitResult(groupID!!, if (alive) delay else -1L, completed) return 0 } }, ) } - completeMissing(plan.profiles.map { it.guid } + plan.failedGuids) + probeIndividually(plan.individualGuids) finish("0") } catch (_: CancellationException) { finish("-1") } catch (error: Throwable) { - if (!finished.get()) { + if (!finished) { LogUtil.e(AppConfig.TAG, "Probe batch failed", error) - finish("-1") + failPending() + finish("0") } } } } private fun startTcpBatch() { - totalProfiles = guids.size val dispatcher = Dispatchers.IO.limitedParallelism(SettingsManager.getRealPingConcurrency()) val jobs = guids.map { guid -> scope.launch(dispatcher) { @@ -111,31 +109,40 @@ class RealPingWorkerService( finish("-1") } + private suspend fun probeIndividually(individualGuids: List) { + val dispatcher = Dispatchers.IO.limitedParallelism(SettingsManager.getRealPingConcurrency()) + individualGuids.map { guid -> + scope.launch(dispatcher) { + emitResult(guid, startRealPing(guid), completed = true) + } + }.joinAll() + } + @Synchronized - private fun completeMissing(allGuids: List) { - allGuids.distinct().forEach { guid -> - if (guid !in completedGuids) emitResult(guid, emittedDelays[guid] ?: -1L, completed = true) + private fun failPending() { + pendingGuids.toList().forEach { guid -> + emitResult(guid, emittedDelays[guid] ?: -1L, completed = true) } } @Synchronized private fun emitResult(guid: String, delay: Long, completed: Boolean) { - if (finished.get()) return + if (finished) return if (emittedDelays[guid] != delay) { emittedDelays[guid] = delay onEvent(RealPingEvent.Result(guid, delay)) } - if (completed && completedGuids.add(guid)) { - val remaining = (totalProfiles - completedGuids.size).coerceAtLeast(0) - onEvent(RealPingEvent.Progress("$remaining / $totalProfiles")) + if (completed) { + pendingGuids.remove(guid) + onEvent(RealPingEvent.Progress("${pendingGuids.size} / ${guids.size}")) } } @Synchronized private fun finish(status: String) { - if (finished.compareAndSet(false, true)) { - onEvent(RealPingEvent.Finish(status)) - } + if (finished) return + finished = true + onEvent(RealPingEvent.Finish(status)) } private fun startTcping(guid: String): Long { @@ -151,4 +158,20 @@ class RealPingWorkerService( } return -1L } + + private fun startRealPing(guid: String): Long { + val config = MmkvManager.decodeServerConfig(guid) ?: return -1L + if (!config.configType.isComplexType() && + config.configType != EConfigType.HYSTERIA2 && + config.configType != EConfigType.WIREGUARD && + config.alpn?.startsWith("h3") != true && + config.server.isNotNullEmpty() && + config.serverPort?.toIntOrNull() != null && + SpeedtestManager.socketConnectTime(config.server.orEmpty(), config.serverPort.orEmpty().toInt(), 1000) <= -1L + ) return -1L + + val configResult = CoreConfigManager.getV2rayConfig4Speedtest(context, guid) + if (!configResult.status) return -1L + return CoreNativeManager.measureOutboundDelay(configResult.content, SettingsManager.getDelayTestUrl()) + } } From b6f4673ed316237fc7538ce6482fd32cc7b79a83 Mon Sep 17 00:00:00 2001 From: Eliot the Cougar Date: Sun, 2 Aug 2026 16:08:45 +0300 Subject: [PATCH 05/10] Use real-delay terminology for probes Reserve speed-test naming for future throughput testing and use HEAD as the default Observatory probe method. --- .../com/v2ray/ang/core/CoreConfigManager.kt | 18 +++++++++--------- .../com/v2ray/ang/core/ProbeConfigBuilder.kt | 4 ++-- .../v2ray/ang/service/RealPingWorkerService.kt | 2 +- 3 files changed, 12 insertions(+), 12 deletions(-) diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt index db9376fc39..cd19a69d67 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt @@ -59,7 +59,7 @@ object CoreConfigManager { * * The core flow is reused, then non-essential sections are removed. */ - fun getV2rayConfig4Speedtest(context: Context, guid: String): ConfigResult { + fun getV2rayConfig4RealDelay(context: Context, guid: String): ConfigResult { try { val configContext = CoreConfigContextBuilder.build(context, guid) ?: return ConfigResult( @@ -70,13 +70,13 @@ object CoreConfigManager { if (configContext.isCustom) { return buildV2rayCustomConfig(configContext) } - return toConfigResult(configContext, buildSpeedtestConfig(configContext)) + return toConfigResult(configContext, buildRealDelayConfig(configContext)) } catch (e: Exception) { - LogUtil.e(AppConfig.TAG, "Failed to get V2ray config for speedtest", e) + LogUtil.e(AppConfig.TAG, "Failed to get V2ray config for real delay", e) return ConfigResult( status = false, guid = guid, - errorMessage = "Failed to get V2ray config: ${e.message ?: e.javaClass.simpleName}" + errorMessage = "Failed to get V2ray config for real delay: ${e.message ?: e.javaClass.simpleName}" ) } } @@ -94,7 +94,7 @@ object CoreConfigManager { } else if (configContext.isCustom) { individualGuids += guid } else { - sources += ProbeConfigBuilder.Source(guid, buildSpeedtestConfig(configContext)) + sources += ProbeConfigBuilder.Source(guid, buildRealDelayConfig(configContext)) } } catch (error: Exception) { LogUtil.e(AppConfig.TAG, "Failed to build probe config for $guid", error) @@ -111,8 +111,8 @@ object CoreConfigManager { ) } - private fun buildSpeedtestConfig(configContext: CoreConfigContext): V2rayConfig = - buildUnifiedConfig(configContext).also(::postProcessForSpeedtest) + private fun buildRealDelayConfig(configContext: CoreConfigContext): V2rayConfig = + buildUnifiedConfig(configContext).also(::postProcessForRealDelay) /** * Build configuration for custom profiles. @@ -463,7 +463,7 @@ object CoreConfigManager { /** * Trim runtime sections that are not needed for latency testing. */ - private fun postProcessForSpeedtest(v2rayConfig: V2rayConfig) { + private fun postProcessForRealDelay(v2rayConfig: V2rayConfig) { v2rayConfig.log.loglevel = MmkvManager.decodeSettingsString(AppConfig.PREF_LOGLEVEL) ?: "warning" v2rayConfig.inbounds.clear() val usesPrimaryBalancer = v2rayConfig.routing.balancers @@ -746,7 +746,7 @@ object CoreConfigManager { } /** - * Remove speed-test runtime sections when the feature is disabled. + * Remove speed-display runtime sections when the feature is disabled. */ private fun applySpeedDisabled(v2rayConfig: V2rayConfig) { if (MmkvManager.decodeSettingsBool(AppConfig.PREF_SPEED_ENABLED) != true) { diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt index 250167d969..9a3ffc4c53 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt @@ -6,7 +6,7 @@ import com.v2ray.ang.dto.ProbeProfile import com.v2ray.ang.dto.V2rayConfig import com.v2ray.ang.util.JsonUtil -/** Combines v2rayNG-generated speed-test configurations into one probe core. */ +/** Combines v2rayNG-generated real-delay configurations into one probe core. */ object ProbeConfigBuilder { data class Source(val guid: String, val config: V2rayConfig) @@ -91,6 +91,6 @@ object ProbeConfigBuilder { } private val OBSERVATORY_STRATEGIES = setOf("leastPing", "leastLoad") - private const val DEFAULT_HTTP_METHOD = "GET" + private const val DEFAULT_HTTP_METHOD = "HEAD" private const val DEFAULT_TIMEOUT = "5s" } diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt index 099b2c029b..d18481b87a 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt @@ -170,7 +170,7 @@ class RealPingWorkerService( SpeedtestManager.socketConnectTime(config.server.orEmpty(), config.serverPort.orEmpty().toInt(), 1000) <= -1L ) return -1L - val configResult = CoreConfigManager.getV2rayConfig4Speedtest(context, guid) + val configResult = CoreConfigManager.getV2rayConfig4RealDelay(context, guid) if (!configResult.status) return -1L return CoreNativeManager.measureOutboundDelay(configResult.content, SettingsManager.getDelayTestUrl()) } From 07692c6f09686fded66fd1bfeac7d6e9d0cb8ab1 Mon Sep 17 00:00:00 2001 From: Eliot the Cougar Date: Sun, 2 Aug 2026 16:57:33 +0300 Subject: [PATCH 06/10] Reconcile probe replacement with notification cancellation Keep cancellation in the batch-test notification while the disposable probe process owns a single active worker. Starting another batch now requests replacement directly, allowing the service to suppress the old worker's completion before handing the new intent to a fresh process instead of emitting an explicit cancellation event that can clear the new UI state. --- .../src/main/java/com/v2ray/ang/service/CoreTestService.kt | 3 --- .../app/src/main/java/com/v2ray/ang/ui/main/MainBottomBar.kt | 3 +-- V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainScreen.kt | 1 - .../app/src/main/java/com/v2ray/ang/ui/main/MainViewModel.kt | 4 +--- 4 files changed, 2 insertions(+), 9 deletions(-) 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 582abd08a6..f0e5e992fd 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 @@ -129,9 +129,6 @@ class CoreTestService : Service() { LogUtil.i(AppConfig.TAG, "CoreTestService cancelling the active batch") replacementRequested = false activeWorker?.cancel() - if (batchFinished.compareAndSet(false, true)) { - MessageHelper.sendMsg2UI(this, AppConfig.MSG_MEASURE_CONFIG_FINISH, "-1") - } activeWorker = null NotificationHelper.stopForeground(this) stopSelf(startId) diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainBottomBar.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainBottomBar.kt index 858e4f6287..ce88845557 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainBottomBar.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainBottomBar.kt @@ -36,7 +36,6 @@ import com.v2ray.ang.ui.compose.colorFabInactiveLight fun MainBottomBar( displayText: String, isRunning: Boolean, - isTesting: Boolean, isDarkTheme: Boolean, onAction: (MainAction) -> Unit ) { @@ -48,7 +47,7 @@ fun MainBottomBar( .fillMaxWidth() .windowInsetsPadding(WindowInsets.navigationBars) .height(64.dp) - .clickable { onAction(if (isTesting) MainAction.CancelTesting else MainAction.TestCurrentServer) }, + .clickable(onClick = { onAction(MainAction.TestCurrentServer) }), color = MaterialTheme.colorScheme.surface, tonalElevation = 0.dp ) { diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainScreen.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainScreen.kt index 37138f2fa3..35b2cbe8bb 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainScreen.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainScreen.kt @@ -236,7 +236,6 @@ fun MainScreen( MainBottomBar( displayText = displayText, isRunning = isRunning, - isTesting = uiState.isTesting, isDarkTheme = isDarkTheme, onAction = onAction ) 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 03b25fca9c..c10aab5022 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 @@ -129,7 +129,6 @@ class MainViewModel( } is MainServiceEvent.MeasureConfigNotify -> { - if (!uiState.value.isTesting) return _uiState.update { it.copy( statusText = dataSource.getString( @@ -672,7 +671,6 @@ class MainViewModel( } fun testAllRealPing(onlyTcp: Boolean = false) { - dataSource.cancelAllPing() val groupId = uiState.value.selectedGroupId val servers = currentServers() dataSource.clearAllTestDelayResults(servers.map { it.guid }) @@ -684,7 +682,7 @@ class MainViewModel( _uiState.update { it.copy( isTesting = true, - statusText = dataSource.getString(R.string.connection_test_testing_tap_to_stop) + statusText = dataSource.getString(R.string.connection_test_testing) ) } viewModelScope.launch(ioDispatcher) { From 75d88e889b3110db26744ec6d984d90a8121f25f Mon Sep 17 00:00:00 2001 From: Eliot the Cougar Date: Sun, 9 Aug 2026 16:44:39 +0300 Subject: [PATCH 07/10] Scale RealDelay probing to large groups --- .../ang/core/CoreConfigContextBuilder.kt | 149 +++++++++++++++--- .../com/v2ray/ang/core/CoreConfigManager.kt | 10 +- .../com/v2ray/ang/core/ProbeConfigBuilder.kt | 11 +- .../java/com/v2ray/ang/dto/RealPingResult.kt | 9 ++ .../com/v2ray/ang/service/CoreTestService.kt | 7 +- .../ang/service/RealPingWorkerService.kt | 41 ++++- .../com/v2ray/ang/ui/main/MainRepository.kt | 15 +- .../com/v2ray/ang/ui/main/MainServiceEvent.kt | 4 +- .../com/v2ray/ang/ui/main/MainViewModel.kt | 67 +++++++- 9 files changed, 264 insertions(+), 49 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/core/CoreConfigContextBuilder.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigContextBuilder.kt index 42ed66980a..33009e5136 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigContextBuilder.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigContextBuilder.kt @@ -4,6 +4,7 @@ import android.content.Context import com.v2ray.ang.AppConfig import com.v2ray.ang.dto.CoreConfigContext import com.v2ray.ang.dto.entities.ProfileItem +import com.v2ray.ang.dto.entities.SubscriptionItem import com.v2ray.ang.enums.BalancerStrategyType import com.v2ray.ang.enums.CoreResolvedType import com.v2ray.ang.enums.EConfigType @@ -23,6 +24,68 @@ import com.v2ray.ang.util.Utils */ object CoreConfigContextBuilder { + /** Lazily decoded profile snapshot shared by every config in one probe batch. */ + class ProbeProfileLookup(requestedGuids: List) { + private val profilesByGuid = linkedMapOf() + private val subscriptionsByGuid = mutableMapOf() + private var profilesByRemarks: Map? = null + private var allProfiles: List? = null + + init { + requestedGuids.distinct().forEach(::loadProfile) + } + + internal fun findByGuid(guid: String): ProfileItem? = + profilesByGuid[guid] ?: loadProfile(guid) + + internal fun findByRemarks(remarks: String?): ProfileItem? { + if (remarks.isNullOrEmpty()) return null + ensureAllProfilesLoaded() + return profilesByRemarks?.get(remarks) + } + + internal fun profiles(): List { + ensureAllProfilesLoaded() + return allProfiles.orEmpty() + } + + internal fun subscription(guid: String): SubscriptionItem? { + if (guid in subscriptionsByGuid) return subscriptionsByGuid[guid] + return MmkvManager.decodeSubscription(guid).also { subscriptionsByGuid[guid] = it } + } + + private fun loadProfile(guid: String): ProfileItem? { + if (guid.isBlank()) return null + return MmkvManager.decodeServerConfig(guid)?.also { profilesByGuid[guid] = it } + } + + private fun ensureAllProfilesLoaded() { + if (allProfiles != null) return + val ordered = mutableListOf() + val seenGuids = mutableSetOf() + MmkvManager.decodeAllServerList().forEach { guid -> + val profile = findByGuid(guid) ?: return@forEach + if (seenGuids.add(guid)) ordered += profile + } + profilesByGuid.forEach { (guid, profile) -> + if (seenGuids.add(guid)) ordered += profile + } + allProfiles = ordered + profilesByRemarks = buildMap { + ordered.forEach { profile -> putIfAbsent(profile.remarks, profile) } + } + } + } + + private fun findProfileByRemarks( + lookup: ProbeProfileLookup?, + remarks: String?, + ): ProfileItem? = if (lookup != null) { + lookup.findByRemarks(remarks) + } else { + SettingsManager.getServerViaRemarks(remarks) + } + /** * Load one profile and produce a fully analyzed context. * @@ -31,22 +94,43 @@ object CoreConfigContextBuilder { fun build(context: Context, guid: String): CoreConfigContext? { val config = MmkvManager.decodeServerConfig(guid) ?: return null + return buildResolved(context, guid, config, lookup = null, includeRouting = true) + } + + /** Build only the outbound dependency graph required by a RealDelay probe. */ + fun buildForProbe( + context: Context, + guid: String, + lookup: ProbeProfileLookup, + ): CoreConfigContext? { + val config = lookup.findByGuid(guid) ?: return null + return buildResolved(context, guid, config, lookup, includeRouting = false) + } + + private fun buildResolved( + context: Context, + guid: String, + config: ProfileItem, + lookup: ProbeProfileLookup?, + includeRouting: Boolean, + ): CoreConfigContext? { + // CUSTOM: return immediately — CoreConfigManager handles this path on its own. if (config.configType == EConfigType.CUSTOM) { return CoreConfigContext(context = context, guid = guid, isCustom = true) } // Step 1: Resolve the main outbound (always tag = TAG_PROXY). - val primaryResolvedOutbound = resolveOutbound(AppConfig.TAG_PROXY, config) ?: run { + val primaryResolvedOutbound = resolveOutbound(AppConfig.TAG_PROXY, config, lookup) ?: run { LogUtil.e(AppConfig.TAG, "Failed to resolve main outbound for '${config.remarks}'") return null } // Step 2: Resolve all non-builtin routing outbound tags. - val routingResolvedOutbounds = resolveRoutingOutbounds() + val routingResolvedOutbounds = if (includeRouting) resolveRoutingOutbounds() else emptyList() val resolvedOutbounds = listOf(primaryResolvedOutbound) + routingResolvedOutbounds - val fallbackResolvedOutbounds = resolveFallbackOutbounds(resolvedOutbounds) - val routingDomainRules = collectRoutingDomainRulesForDns() + val fallbackResolvedOutbounds = resolveFallbackOutbounds(resolvedOutbounds, lookup) + val routingDomainRules = if (includeRouting) collectRoutingDomainRulesForDns() else emptyList() return CoreConfigContext( context = context, @@ -61,25 +145,29 @@ object CoreConfigContextBuilder { * * Custom profiles are ignored at this stage and produce no entry. */ - private fun resolveOutbound(tag: String, profile: ProfileItem): CoreConfigContext.ResolvedOutbound? { + private fun resolveOutbound( + tag: String, + profile: ProfileItem, + lookup: ProbeProfileLookup? = null, + ): CoreConfigContext.ResolvedOutbound? { if (profile.configType == EConfigType.CUSTOM) { return null } val (resolvedProfiles, resolvedType) = when (profile.configType) { EConfigType.POLICYGROUP -> Pair( - resolvePolicyGroupProfiles(profile), + resolvePolicyGroupProfiles(profile, lookup), CoreResolvedType.POLICYGROUP, ) EConfigType.PROXYCHAIN -> { - val chainProfiles = resolveProxyChainProfiles(profile) + val chainProfiles = resolveProxyChainProfiles(profile, lookup) val type = if (chainProfiles.size <= 1) CoreResolvedType.NORMAL else CoreResolvedType.PROXYCHAIN Pair(chainProfiles, type) } else -> { - val chainProfiles = resolveProxyChainProfilesFromGroup(profile) + val chainProfiles = resolveProxyChainProfilesFromGroup(profile, lookup) val type = if (chainProfiles.size <= 1) CoreResolvedType.NORMAL else CoreResolvedType.PROXYCHAIN Pair(chainProfiles, type) } @@ -141,12 +229,14 @@ object CoreConfigContextBuilder { return resolvedOutbounds } - private fun resolvePolicyGroupProfiles(config: ProfileItem): List { + private fun resolvePolicyGroupProfiles( + config: ProfileItem, + lookup: ProbeProfileLookup?, + ): List { try { - val serverList = MmkvManager.decodeAllServerList() - return serverList - .asSequence() - .mapNotNull { id -> MmkvManager.decodeServerConfig(id) } + val profiles = lookup?.profiles() ?: MmkvManager.decodeAllServerList() + .mapNotNull(MmkvManager::decodeServerConfig) + return profiles.asSequence() .filter { profile -> val subscriptionId = config.policyGroupSubscriptionId if (subscriptionId.isNullOrBlank()) { @@ -177,7 +267,10 @@ object CoreConfigContextBuilder { } } - private fun resolveProxyChainProfiles(config: ProfileItem): List { + private fun resolveProxyChainProfiles( + config: ProfileItem, + lookup: ProbeProfileLookup?, + ): List { if (config.proxyChainProfiles.isNullOrBlank()) { return listOf(config) } @@ -185,7 +278,7 @@ object CoreConfigContextBuilder { try { return config.proxyChainProfiles.orEmpty().split(",") .asSequence() - .mapNotNull { remark -> SettingsManager.getServerViaRemarks(remark) } + .mapNotNull { remark -> findProfileByRemarks(lookup, remark) } .filter { it.server.isNotNullEmpty() } .filter { Utils.isPureIpAddress(it.server!!) || Utils.isValidUrl(it.server!!) } .filter { !it.configType.isComplexType() } @@ -202,17 +295,26 @@ object CoreConfigContextBuilder { * * When no chain is available, return a single-node result. */ - private fun resolveProxyChainProfilesFromGroup(config: ProfileItem): List { + private fun resolveProxyChainProfilesFromGroup( + config: ProfileItem, + lookup: ProbeProfileLookup?, + ): List { if (config.subscriptionId.isEmpty()) { return listOf(config) } try { - val subItem = MmkvManager.decodeSubscription(config.subscriptionId) ?: return listOf(config) + val subItem = if (lookup != null) { + lookup.subscription(config.subscriptionId) + } else { + MmkvManager.decodeSubscription(config.subscriptionId) + } ?: return listOf(config) val resolved = mutableListOf() - SettingsManager.getServerViaRemarks(subItem.nextProfile)?.let { resolved.add(it) } + findProfileByRemarks(lookup, subItem.nextProfile) + ?.let { resolved.add(it) } resolved.add(config) - SettingsManager.getServerViaRemarks(subItem.prevProfile)?.let { resolved.add(it) } + findProfileByRemarks(lookup, subItem.prevProfile) + ?.let { resolved.add(it) } return resolved } catch (e: Exception) { LogUtil.e(AppConfig.TAG, "Failed to resolve proxy chain from group for '${config.remarks}'", e) @@ -255,7 +357,10 @@ object CoreConfigContextBuilder { * * Fallback targets must not overlap with already resolved tags or builtin tags. */ - private fun resolveFallbackOutbounds(resolvedOutbounds: List): List { + private fun resolveFallbackOutbounds( + resolvedOutbounds: List, + lookup: ProbeProfileLookup?, + ): List { return resolvedOutbounds .asSequence() .filter { it.resolvedType == CoreResolvedType.POLICYGROUP } @@ -264,9 +369,9 @@ object CoreConfigContextBuilder { .filter { it !in AppConfig.BUILTIN_OUTBOUND_TAGS && resolvedOutbounds.none { outbound -> outbound.tag == it } } .distinct() .mapNotNull { tag -> - SettingsManager.getServerViaRemarks(tag) + findProfileByRemarks(lookup, tag) ?.takeUnless { it.configType == EConfigType.CUSTOM || it.configType == EConfigType.POLICYGROUP } - ?.let { resolveOutbound(tag, it) } + ?.let { resolveOutbound(tag, it, lookup) } } .toList() } diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt index cd19a69d67..6487e95010 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt @@ -86,9 +86,11 @@ object CoreConfigManager { val sources = mutableListOf() val individualGuids = mutableListOf() val failedGuids = mutableListOf() - guids.distinct().forEach { guid -> + val distinctGuids = guids.distinct() + val profileLookup = CoreConfigContextBuilder.ProbeProfileLookup(distinctGuids) + distinctGuids.forEach { guid -> try { - val configContext = CoreConfigContextBuilder.build(context, guid) + val configContext = CoreConfigContextBuilder.buildForProbe(context, guid, profileLookup) if (configContext == null) { failedGuids += guid } else if (configContext.isCustom) { @@ -105,9 +107,11 @@ object CoreConfigManager { sources = sources, destination = SettingsManager.getDelayTestUrl(), ) + val emptyProfiles = plan.profiles.filter { it.outboundTags.isEmpty() } return plan.copy( + profiles = plan.profiles.filter { it.outboundTags.isNotEmpty() }, individualGuids = (individualGuids + plan.individualGuids).distinct(), - failedGuids = failedGuids, + failedGuids = (failedGuids + emptyProfiles.map { it.guid }).distinct(), ) } diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt index 9a3ffc4c53..0cbb99c3f4 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt @@ -15,8 +15,6 @@ object ProbeConfigBuilder { val balancers = mutableListOf() val profiles = mutableListOf() val individualGuids = mutableListOf() - var httpMethod = DEFAULT_HTTP_METHOD - var timeout = DEFAULT_TIMEOUT sources.forEachIndexed { index, source -> val namespace = "probe-$index-" @@ -55,11 +53,6 @@ object ProbeConfigBuilder { balancers += probeBalancer profiles += ProbeProfile(source.guid, outboundTags, probeBalancer.tag) - if (strategyType == "leastLoad") { - val pingConfig = (source.config.burstObservatory as V2rayConfig.BurstObservatoryObject).pingConfig - httpMethod = pingConfig.httpMethod ?: DEFAULT_HTTP_METHOD - timeout = pingConfig.timeout ?: DEFAULT_TIMEOUT - } } val routing = mutableMapOf( @@ -76,10 +69,10 @@ object ProbeConfigBuilder { "subjectSelector" to emptyList(), "pingConfig" to mapOf( "destination" to destination, - "httpMethod" to httpMethod, + "httpMethod" to DEFAULT_HTTP_METHOD, "interval" to "1h", "sampling" to 1, - "timeout" to timeout, + "timeout" to DEFAULT_TIMEOUT, ), ), ) 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 0000000000..ab2b8614df --- /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 RealDelay result delivered from the probe 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 f0e5e992fd..bad83fe866 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 @@ -13,6 +13,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 @@ -150,7 +151,11 @@ class CoreTestService : Service() { is RealPingEvent.Result -> { MmkvManager.encodeServerTestDelayMillis(event.guid, event.delayMillis) - MessageHelper.sendMsg2UI(this, AppConfig.MSG_MEASURE_CONFIG_SUCCESS, event.guid) + MessageHelper.sendMsg2UI( + this, + AppConfig.MSG_MEASURE_CONFIG_SUCCESS, + RealPingResult(event.guid, event.delayMillis), + ) } is RealPingEvent.Finish -> { diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt index d18481b87a..1d5d8b1554 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt @@ -1,6 +1,7 @@ package com.v2ray.ang.service import android.content.Context +import android.os.SystemClock import com.v2ray.ang.AppConfig import com.v2ray.ang.core.CoreConfigManager import com.v2ray.ang.core.CoreNativeManager @@ -38,6 +39,9 @@ class RealPingWorkerService( private var finished = false private val emittedDelays = mutableMapOf() private val pendingGuids = guids.toMutableSet() + private var completedWorkUnits = 0 + private var totalWorkUnits = guids.size + private var lastProgressAt = 0L fun start() { if (onlyTcp) { @@ -47,10 +51,14 @@ class RealPingWorkerService( scope.launch { try { val plan = CoreConfigManager.getProbePlan(context, guids) - plan.failedGuids.forEach { emitResult(it, -1L, completed = true) } + val probeCount = plan.profiles.sumOf { it.outboundTags.size } + setTotalWorkUnits(probeCount + plan.individualGuids.size + plan.failedGuids.size) + plan.failedGuids.forEach { + emitResult(it, -1L, completed = true) + completeWorkUnit() + } if (plan.profiles.isNotEmpty()) { val concurrency = SettingsManager.getRealPingConcurrency() - val probeCount = plan.profiles.sumOf { it.outboundTags.size } LogUtil.i( AppConfig.TAG, "Starting $probeCount real-delay probes for ${plan.profiles.size} profiles with limit $concurrency", @@ -67,6 +75,7 @@ class RealPingWorkerService( completed: Boolean, ): Long { emitResult(groupID!!, if (alive) delay else -1L, completed) + completeWorkUnit() return 0 } }, @@ -91,6 +100,7 @@ class RealPingWorkerService( val jobs = guids.map { guid -> scope.launch(dispatcher) { emitResult(guid, startTcping(guid), completed = true) + completeWorkUnit() } } scope.launch { @@ -114,6 +124,7 @@ class RealPingWorkerService( individualGuids.map { guid -> scope.launch(dispatcher) { emitResult(guid, startRealPing(guid), completed = true) + completeWorkUnit() } }.joinAll() } @@ -134,10 +145,30 @@ class RealPingWorkerService( } if (completed) { pendingGuids.remove(guid) - onEvent(RealPingEvent.Progress("${pendingGuids.size} / ${guids.size}")) } } + @Synchronized + private fun setTotalWorkUnits(total: Int) { + totalWorkUnits = total.coerceAtLeast(1) + completedWorkUnits = 0 + lastProgressAt = 0L + emitProgress(force = true) + } + + @Synchronized + private fun completeWorkUnit() { + completedWorkUnits = (completedWorkUnits + 1).coerceAtMost(totalWorkUnits) + emitProgress(force = completedWorkUnits == totalWorkUnits) + } + + private fun emitProgress(force: Boolean) { + val now = SystemClock.elapsedRealtime() + if (!force && now - lastProgressAt < PROGRESS_UPDATE_INTERVAL_MS) return + lastProgressAt = now + onEvent(RealPingEvent.Progress("$completedWorkUnits / $totalWorkUnits")) + } + @Synchronized private fun finish(status: String) { if (finished) return @@ -174,4 +205,8 @@ class RealPingWorkerService( if (!configResult.status) return -1L return CoreNativeManager.measureOutboundDelay(configResult.content, SettingsManager.getDelayTestUrl()) } + + private companion object { + const val PROGRESS_UPDATE_INTERVAL_MS = 100L + } } 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 d1e0e0fb7b..30bcf98af3 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 @@ -8,12 +8,14 @@ import androidx.core.content.ContextCompat import com.v2ray.ang.AngApplication import com.v2ray.ang.AppConfig import com.v2ray.ang.R +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 import com.v2ray.ang.dto.entities.ServerAffiliationInfo import com.v2ray.ang.dto.entities.SubscriptionCache import com.v2ray.ang.dto.entities.SubscriptionItem +import com.v2ray.ang.extension.serializable import com.v2ray.ang.handler.AngConfigManager import com.v2ray.ang.handler.AppLocaleManager import com.v2ray.ang.handler.MmkvManager @@ -39,12 +41,19 @@ class MainRepository( private val _mainServiceEvent = MutableSharedFlow( replay = 0, - extraBufferCapacity = 64, + // A large policy group can finish an entire concurrency wave between + // main-thread dispatches. Keep every incremental result until the + // ViewModel coalesces it into one row update. + extraBufferCapacity = SERVICE_EVENT_BUFFER_CAPACITY, onBufferOverflow = BufferOverflow.DROP_OLDEST ) override val mainServiceEvent: SharedFlow = _mainServiceEvent.asSharedFlow() + private companion object { + const val SERVICE_EVENT_BUFFER_CAPACITY = 2048 + } + private val serviceReceiver = object : BroadcastReceiver() { override fun onReceive(context: Context?, intent: Intent?) { val safeIntent = intent ?: return @@ -61,7 +70,9 @@ class MainRepository( safeIntent.getStringExtra("content").orEmpty() ) - AppConfig.MSG_MEASURE_CONFIG_SUCCESS -> MainServiceEvent.MeasureConfigSuccess + AppConfig.MSG_MEASURE_CONFIG_SUCCESS -> safeIntent + .serializable("content") + ?.let(MainServiceEvent::MeasureConfigSuccess) AppConfig.MSG_MEASURE_CONFIG_NOTIFY -> MainServiceEvent.MeasureConfigNotify( safeIntent.getStringExtra("content").orEmpty() ) 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 105092c7ac..45f035fd64 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,5 +1,7 @@ package com.v2ray.ang.ui.main +import com.v2ray.ang.dto.RealPingResult + sealed class MainServiceEvent { data object StateRunning : MainServiceEvent() data object StateNotRunning : MainServiceEvent() @@ -7,7 +9,7 @@ sealed class MainServiceEvent { data class StateStartFailure(val errorMessage: String) : MainServiceEvent() data object StateStopSuccess : MainServiceEvent() data class MeasureDelaySuccess(val content: String) : MainServiceEvent() - data object MeasureConfigSuccess : MainServiceEvent() + data class MeasureConfigSuccess(val result: RealPingResult) : MainServiceEvent() data class MeasureConfigNotify(val progress: String) : MainServiceEvent() data class MeasureConfigFinish(val finishedCount: 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 c10aab5022..cf589ec848 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 @@ -8,6 +8,7 @@ import com.v2ray.ang.AppConfig import com.v2ray.ang.R 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 @@ -76,6 +77,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() @Volatile private var testingGroupId: String? = null @@ -120,13 +123,7 @@ class MainViewModel( _uiState.update { it.copy(statusText = event.content) } } - MainServiceEvent.MeasureConfigSuccess -> { - viewModelScope.launch(ioDispatcher) { - val gid = testingGroupId ?: uiState.value.selectedGroupId - cacheMutex.withLock { groupDataCache.remove(gid) } - updateGroupUi(gid, loadGroup(gid, forceRefresh = true)) - } - } + is MainServiceEvent.MeasureConfigSuccess -> queueTestResult(event.result) is MainServiceEvent.MeasureConfigNotify -> { _uiState.update { @@ -140,9 +137,46 @@ class MainViewModel( } is MainServiceEvent.MeasureConfigFinish -> { - onTestsFinished() + testResultFlushJob?.cancel() + testResultFlushJob = viewModelScope.launch { + flushPendingTestResults() + onTestsFinished() + } + } + } + } + + private fun queueTestResult(result: RealPingResult) { + pendingTestResults[result.guid] = result.delayMillis + if (testResultFlushJob?.isActive == true) return + testResultFlushJob = viewModelScope.launch { + delay(TEST_RESULT_FLUSH_INTERVAL_MS) + flushPendingTestResults() + } + } + + private suspend fun flushPendingTestResults() { + if (pendingTestResults.isEmpty()) return + val updates = pendingTestResults.toMap() + pendingTestResults.clear() + val groupId = testingGroupId ?: uiState.value.selectedGroupId + val applyUpdates: (List) -> List = { servers -> + servers.map { server -> + val delayMillis = updates[server.guid] + if (delayMillis == null || delayMillis == server.testDelayMillis) { + server + } else { + server.copy( + testDelayMillis = delayMillis, + testDelayString = if (delayMillis == 0L) "" else "${delayMillis}ms", + ) + } } } + cacheMutex.withLock { + groupDataCache[groupId]?.let { groupDataCache[groupId] = applyUpdates(it) } + } + mutableServersForGroup(groupId).update(applyUpdates) } // ---------- Public state accessors ---------- @@ -661,6 +695,9 @@ class MainViewModel( // ---------- Testing ---------- fun cancelAllPing() { dataSource.cancelAllPing() + testResultFlushJob?.cancel() + testResultFlushJob = null + pendingTestResults.clear() testingGroupId = null _uiState.update { it.copy( @@ -673,11 +710,20 @@ class MainViewModel( fun testAllRealPing(onlyTcp: Boolean = false) { val groupId = uiState.value.selectedGroupId val servers = currentServers() + testResultFlushJob?.cancel() + testResultFlushJob = null + pendingTestResults.clear() dataSource.clearAllTestDelayResults(servers.map { it.guid }) if (servers.isEmpty()) { _uiState.update { it.copy(isTesting = false) } return } + mutableServersForGroup(groupId).update { current -> + current.map { server -> + if (server.testDelayMillis == 0L) server + else server.copy(testDelayMillis = 0L, testDelayString = "") + } + } testingGroupId = groupId _uiState.update { it.copy( @@ -762,6 +808,7 @@ class MainViewModel( selectedGroupLoadJob?.cancel() reloadJob?.cancel() filterJob?.cancel() + testResultFlushJob?.cancel() cancelAllPing() dataSource.close() super.onCleared() @@ -777,4 +824,8 @@ class MainViewModel( throw IllegalArgumentException("Unknown ViewModel class") } } + + private companion object { + const val TEST_RESULT_FLUSH_INTERVAL_MS = 50L + } } From d30745ccf56ac629968695d93ba935838ba8a5e5 Mon Sep 17 00:00:00 2001 From: Eliot the Cougar Date: Sun, 9 Aug 2026 19:52:09 +0300 Subject: [PATCH 08/10] Simplify and harden observatory probing --- .../com/v2ray/ang/core/CoreConfigManager.kt | 6 +- .../com/v2ray/ang/core/ProbeConfigBuilder.kt | 107 ++++++++---- .../ang/service/RealPingWorkerService.kt | 163 ++++++++++++------ 3 files changed, 185 insertions(+), 91 deletions(-) diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt index 6487e95010..4b202ac9ed 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt @@ -107,11 +107,9 @@ object CoreConfigManager { sources = sources, destination = SettingsManager.getDelayTestUrl(), ) - val emptyProfiles = plan.profiles.filter { it.outboundTags.isEmpty() } return plan.copy( - profiles = plan.profiles.filter { it.outboundTags.isNotEmpty() }, - individualGuids = (individualGuids + plan.individualGuids).distinct(), - failedGuids = (failedGuids + emptyProfiles.map { it.guid }).distinct(), + individualGuids = individualGuids + plan.individualGuids, + failedGuids = failedGuids, ) } diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt index 0cbb99c3f4..1b8d6433cb 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt @@ -17,42 +17,18 @@ object ProbeConfigBuilder { val individualGuids = mutableListOf() sources.forEachIndexed { index, source -> - val namespace = "probe-$index-" - val tagMap = source.config.outbounds.associate { it.tag to "$namespace${it.tag}" } - val primaryBalancer = source.config.routing.balancers - ?.firstOrNull { it.tag == AppConfig.TAG_BALANCER } - val strategyType = primaryBalancer?.strategy?.type - - if (primaryBalancer != null && strategyType !in OBSERVATORY_STRATEGIES) { - individualGuids += source.guid - return@forEachIndexed - } - - source.config.outbounds.forEach { outbound -> - outbound.tag = tagMap.getValue(outbound.tag) - outbound.streamSettings?.sockopt?.dialerProxy?.let { dialerProxy -> - outbound.streamSettings?.sockopt?.dialerProxy = tagMap.getValue(dialerProxy) - } - outbounds += outbound + val prepared = try { + prepareSource(source, index) + } catch (_: Exception) { + null } - - if (primaryBalancer == null) { - profiles += ProbeProfile(source.guid, listOf(tagMap.getValue(AppConfig.TAG_PROXY))) + if (prepared == null) { + individualGuids += source.guid return@forEachIndexed } - - val outboundTags = tagMap - .filterKeys { tag -> primaryBalancer.selector.any(tag::startsWith) } - .values - .toList() - val probeBalancer = primaryBalancer.copy( - tag = "$namespace${primaryBalancer.tag}", - selector = primaryBalancer.selector.map { "$namespace$it" }, - fallbackTag = primaryBalancer.fallbackTag?.let(tagMap::getValue), - ) - balancers += probeBalancer - profiles += ProbeProfile(source.guid, outboundTags, probeBalancer.tag) - + outbounds += prepared.outbounds + prepared.balancer?.let(balancers::add) + profiles += prepared.profile } val routing = mutableMapOf( @@ -83,7 +59,70 @@ object ProbeConfigBuilder { ) } - private val OBSERVATORY_STRATEGIES = setOf("leastPing", "leastLoad") + /** Validate and rewrite one source completely before exposing any part of it to the batch. */ + private fun prepareSource(source: Source, index: Int): PreparedSource? { + val sourceOutbounds = source.config.outbounds + val namespace = "probe-$index-" + val tagMap = sourceOutbounds.associate { it.tag to "$namespace${it.tag}" } + val primaryBalancer = source.config.routing.balancers + ?.firstOrNull { it.tag == AppConfig.TAG_BALANCER } + val strategyType = primaryBalancer?.strategy?.type?.lowercase() + if (primaryBalancer != null && + (strategyType !in SUPPORTED_BALANCER_STRATEGIES || primaryBalancer.fallbackTag != null) + ) { + return null + } + + // Resolve every reference before mutating anything, so a malformed source + // cannot leave part of itself in the shared configuration. + val mappedDialerProxies = sourceOutbounds.map { outbound -> + outbound.streamSettings?.sockopt?.dialerProxy?.let { tagMap[it] ?: return null } + } + + val profile: ProbeProfile + val probeBalancer: V2rayConfig.RoutingBean.BalancerBean? + if (primaryBalancer == null) { + val proxyTag = tagMap[AppConfig.TAG_PROXY] ?: return null + profile = ProbeProfile(source.guid, listOf(proxyTag)) + probeBalancer = null + } else { + val selectors = primaryBalancer.selector + if (selectors.isEmpty() || selectors.any(String::isBlank)) return null + val outboundTags = tagMap + .filterKeys { tag -> selectors.any(tag::startsWith) } + .values + .toList() + if (outboundTags.isEmpty()) return null + probeBalancer = primaryBalancer.copy( + tag = "$namespace${primaryBalancer.tag}", + selector = selectors.map { "$namespace$it" }, + ) + profile = ProbeProfile( + guid = source.guid, + outboundTags = outboundTags, + balancerTag = probeBalancer.tag, + ) + } + + sourceOutbounds.forEachIndexed { outboundIndex, outbound -> + outbound.tag = tagMap.getValue(outbound.tag) + mappedDialerProxies[outboundIndex]?.let { mapped -> + outbound.streamSettings?.sockopt?.dialerProxy = mapped + } + } + return PreparedSource(sourceOutbounds, probeBalancer, profile) + } + + private data class PreparedSource( + val outbounds: List, + val balancer: V2rayConfig.RoutingBean.BalancerBean?, + val profile: ProbeProfile, + ) + + private val SUPPORTED_BALANCER_STRATEGIES = setOf( + "leastping", + "leastload", + ) private const val DEFAULT_HTTP_METHOD = "HEAD" private const val DEFAULT_TIMEOUT = "5s" } diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt index 1d5d8b1554..8ee5eb9eed 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt @@ -5,6 +5,7 @@ import android.os.SystemClock import com.v2ray.ang.AppConfig import com.v2ray.ang.core.CoreConfigManager import com.v2ray.ang.core.CoreNativeManager +import com.v2ray.ang.dto.ProbePlan import com.v2ray.ang.dto.RealPingEvent import com.v2ray.ang.enums.EConfigType import com.v2ray.ang.extension.isComplexType @@ -18,7 +19,9 @@ import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineName import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers -import kotlinx.coroutines.Job +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.currentCoroutineContext +import kotlinx.coroutines.ensureActive import kotlinx.coroutines.joinAll import kotlinx.coroutines.launch import libv2ray.Libv2ray @@ -32,15 +35,15 @@ class RealPingWorkerService( private val onEvent: (RealPingEvent) -> Unit = {}, ) { private val guids = guids.distinct() - private val job = Job() + private val job = SupervisorJob() private val scope = CoroutineScope(job + Dispatchers.IO + CoroutineName("ProbeBatch")) private val controller = Libv2ray.newProbeController() @Volatile private var finished = false private val emittedDelays = mutableMapOf() - private val pendingGuids = guids.toMutableSet() private var completedWorkUnits = 0 private var totalWorkUnits = guids.size + private val remainingWorkUnits = guids.associateWith { 1 }.toMutableMap() private var lastProgressAt = 0L fun start() { @@ -52,36 +55,16 @@ class RealPingWorkerService( try { val plan = CoreConfigManager.getProbePlan(context, guids) val probeCount = plan.profiles.sumOf { it.outboundTags.size } - setTotalWorkUnits(probeCount + plan.individualGuids.size + plan.failedGuids.size) - plan.failedGuids.forEach { - emitResult(it, -1L, completed = true) - completeWorkUnit() - } + setWorkUnits(plan) + val concurrency = SettingsManager.getRealPingConcurrency() if (plan.profiles.isNotEmpty()) { - val concurrency = SettingsManager.getRealPingConcurrency() LogUtil.i( AppConfig.TAG, "Starting $probeCount real-delay probes for ${plan.profiles.size} profiles with limit $concurrency", ) - controller.probe( - plan.content, - JsonUtil.toJson(plan.profiles), - concurrency, - object : ProbeHandler { - override fun onProbeResult( - groupID: String?, - delay: Long, - alive: Boolean, - completed: Boolean, - ): Long { - emitResult(groupID!!, if (alive) delay else -1L, completed) - completeWorkUnit() - return 0 - } - }, - ) } - probeIndividually(plan.individualGuids) + runPlan(plan, concurrency) + failPending() finish("0") } catch (_: CancellationException) { finish("-1") @@ -99,8 +82,8 @@ class RealPingWorkerService( val dispatcher = Dispatchers.IO.limitedParallelism(SettingsManager.getRealPingConcurrency()) val jobs = guids.map { guid -> scope.launch(dispatcher) { - emitResult(guid, startTcping(guid), completed = true) - completeWorkUnit() + emitResult(guid, safelyProbe(guid, ::startTcping)) + completeWork(guid, profileCompleted = true) } } scope.launch { @@ -119,46 +102,121 @@ class RealPingWorkerService( finish("-1") } - private suspend fun probeIndividually(individualGuids: List) { - val dispatcher = Dispatchers.IO.limitedParallelism(SettingsManager.getRealPingConcurrency()) + private suspend fun runPlan(plan: ProbePlan, concurrency: Int) { + plan.failedGuids.forEach { guid -> + emitResult(guid, -1L) + completeWork(guid, profileCompleted = true) + } + probeBatch(plan, concurrency) + probeIndividually(plan.individualGuids, concurrency) + } + + private suspend fun probeIndividually(individualGuids: List, concurrency: Int) { + val dispatcher = Dispatchers.IO.limitedParallelism(concurrency) individualGuids.map { guid -> scope.launch(dispatcher) { - emitResult(guid, startRealPing(guid), completed = true) - completeWorkUnit() + emitResult(guid, safelyProbe(guid, ::startRealPing)) + completeWork(guid, profileCompleted = true) } }.joinAll() } + private suspend fun probeBatch(plan: ProbePlan, concurrency: Int) { + if (plan.profiles.isEmpty()) return + try { + controller.probe( + plan.content, + JsonUtil.toJson(plan.profiles), + concurrency, + object : ProbeHandler { + override fun onProbeResult( + groupID: String?, + delay: Long, + completed: Boolean, + ) { + val guid = groupID ?: return + emitResult(guid, delay) + completeWork(guid, profileCompleted = completed) + } + }, + ) + } catch (error: CancellationException) { + throw error + } catch (error: Exception) { + currentCoroutineContext().ensureActive() + val retryGuids = plan.profiles.map { it.guid }.filter(::isPending) + LogUtil.w( + AppConfig.TAG, + "Shared probe core rejected ${retryGuids.size} profiles; isolating the failing profile", + error, + ) + retryProbeGuids(retryGuids, concurrency) + } + } + + /** Binary isolation keeps one malformed Xray config from degrading every valid profile. */ + private suspend fun retryProbeGuids(retryGuids: List, concurrency: Int) { + currentCoroutineContext().ensureActive() + val activeGuids = retryGuids.filter(::isPending) + if (activeGuids.isEmpty()) return + if (activeGuids.size == 1) { + probeIndividually(activeGuids, concurrency) + return + } + val halves = activeGuids.chunked((activeGuids.size + 1) / 2) + halves.forEach { half -> + currentCoroutineContext().ensureActive() + val retryPlan = try { + CoreConfigManager.getProbePlan(context, half) + } catch (error: Exception) { + LogUtil.w(AppConfig.TAG, "Failed to rebuild ${half.size} isolated probe profiles", error) + retryProbeGuids(half, concurrency) + return@forEach + } + runPlan(retryPlan, concurrency) + } + } + @Synchronized private fun failPending() { - pendingGuids.toList().forEach { guid -> - emitResult(guid, emittedDelays[guid] ?: -1L, completed = true) + remainingWorkUnits.filterValues { it > 0 }.keys.toList().forEach { guid -> + emitResult(guid, emittedDelays[guid] ?: -1L) + completeWork(guid, profileCompleted = true) } } @Synchronized - private fun emitResult(guid: String, delay: Long, completed: Boolean) { + private fun isPending(guid: String): Boolean = remainingWorkUnits[guid]?.let { it > 0 } == true + + @Synchronized + private fun emitResult(guid: String, delay: Long) { if (finished) return if (emittedDelays[guid] != delay) { emittedDelays[guid] = delay onEvent(RealPingEvent.Result(guid, delay)) } - if (completed) { - pendingGuids.remove(guid) - } } @Synchronized - private fun setTotalWorkUnits(total: Int) { - totalWorkUnits = total.coerceAtLeast(1) + private fun setWorkUnits(plan: ProbePlan) { + remainingWorkUnits.clear() + guids.forEach { remainingWorkUnits[it] = 1 } + plan.profiles.forEach { profile -> + remainingWorkUnits[profile.guid] = profile.outboundTags.size.coerceAtLeast(1) + } + totalWorkUnits = remainingWorkUnits.values.sum().coerceAtLeast(1) completedWorkUnits = 0 lastProgressAt = 0L emitProgress(force = true) } @Synchronized - private fun completeWorkUnit() { - completedWorkUnits = (completedWorkUnits + 1).coerceAtMost(totalWorkUnits) + private fun completeWork(guid: String, profileCompleted: Boolean) { + val remaining = remainingWorkUnits[guid] ?: return + if (remaining <= 0) return + val completed = if (profileCompleted) remaining else 1 + remainingWorkUnits[guid] = remaining - completed + completedWorkUnits = (completedWorkUnits + completed).coerceAtMost(totalWorkUnits) emitProgress(force = completedWorkUnits == totalWorkUnits) } @@ -191,21 +249,20 @@ class RealPingWorkerService( } private fun startRealPing(guid: String): Long { - val config = MmkvManager.decodeServerConfig(guid) ?: return -1L - if (!config.configType.isComplexType() && - config.configType != EConfigType.HYSTERIA2 && - config.configType != EConfigType.WIREGUARD && - config.alpn?.startsWith("h3") != true && - config.server.isNotNullEmpty() && - config.serverPort?.toIntOrNull() != null && - SpeedtestManager.socketConnectTime(config.server.orEmpty(), config.serverPort.orEmpty().toInt(), 1000) <= -1L - ) return -1L - val configResult = CoreConfigManager.getV2rayConfig4RealDelay(context, guid) if (!configResult.status) return -1L return CoreNativeManager.measureOutboundDelay(configResult.content, SettingsManager.getDelayTestUrl()) } + private fun safelyProbe(guid: String, probe: (String) -> Long): Long = try { + probe(guid) + } catch (error: CancellationException) { + throw error + } catch (error: Exception) { + LogUtil.e(AppConfig.TAG, "Probe failed for $guid", error) + -1L + } + private companion object { const val PROGRESS_UPDATE_INTERVAL_MS = 100L } From 2f9b560b2cc08be371985103e9c682b60877979d Mon Sep 17 00:00:00 2001 From: Eliot the Cougar Date: Sun, 9 Aug 2026 22:06:31 +0300 Subject: [PATCH 09/10] Refine observatory delay probing --- .../ang/core/CoreConfigContextBuilder.kt | 15 +++--- .../com/v2ray/ang/core/CoreConfigManager.kt | 2 +- .../com/v2ray/ang/core/CoreNativeManager.kt | 2 +- .../com/v2ray/ang/core/ProbeConfigBuilder.kt | 4 +- .../com/v2ray/ang/service/CoreTestService.kt | 2 +- .../ang/service/RealPingWorkerService.kt | 2 +- .../ang/service/SubscriptionUpdateService.kt | 47 ++++++++++--------- .../com/v2ray/ang/ui/main/MainRepository.kt | 4 +- .../com/v2ray/ang/ui/main/MainViewModel.kt | 11 +++-- 9 files changed, 45 insertions(+), 44 deletions(-) diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigContextBuilder.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigContextBuilder.kt index 33009e5136..7a38957bb8 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigContextBuilder.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigContextBuilder.kt @@ -25,31 +25,31 @@ import com.v2ray.ang.util.Utils object CoreConfigContextBuilder { /** Lazily decoded profile snapshot shared by every config in one probe batch. */ - class ProbeProfileLookup(requestedGuids: List) { + internal class ProbeProfileLookup(requestedGuids: List) { private val profilesByGuid = linkedMapOf() private val subscriptionsByGuid = mutableMapOf() private var profilesByRemarks: Map? = null private var allProfiles: List? = null init { - requestedGuids.distinct().forEach(::loadProfile) + requestedGuids.forEach(::loadProfile) } - internal fun findByGuid(guid: String): ProfileItem? = + fun findByGuid(guid: String): ProfileItem? = profilesByGuid[guid] ?: loadProfile(guid) - internal fun findByRemarks(remarks: String?): ProfileItem? { + fun findByRemarks(remarks: String?): ProfileItem? { if (remarks.isNullOrEmpty()) return null ensureAllProfilesLoaded() return profilesByRemarks?.get(remarks) } - internal fun profiles(): List { + fun profiles(): List { ensureAllProfilesLoaded() return allProfiles.orEmpty() } - internal fun subscription(guid: String): SubscriptionItem? { + fun subscription(guid: String): SubscriptionItem? { if (guid in subscriptionsByGuid) return subscriptionsByGuid[guid] return MmkvManager.decodeSubscription(guid).also { subscriptionsByGuid[guid] = it } } @@ -98,7 +98,7 @@ object CoreConfigContextBuilder { } /** Build only the outbound dependency graph required by a RealDelay probe. */ - fun buildForProbe( + internal fun buildForProbe( context: Context, guid: String, lookup: ProbeProfileLookup, @@ -114,7 +114,6 @@ object CoreConfigContextBuilder { lookup: ProbeProfileLookup?, includeRouting: Boolean, ): CoreConfigContext? { - // CUSTOM: return immediately — CoreConfigManager handles this path on its own. if (config.configType == EConfigType.CUSTOM) { return CoreConfigContext(context = context, guid = guid, isCustom = true) diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt index 4b202ac9ed..9fe768e4f9 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt @@ -82,7 +82,7 @@ object CoreConfigManager { } /** Builds one isolated Xray configuration for a complete UI delay-test batch. */ - fun getProbePlan(context: Context, guids: List): ProbePlan { + internal fun getProbePlan(context: Context, guids: List): ProbePlan { val sources = mutableListOf() val individualGuids = mutableListOf() val failedGuids = mutableListOf() diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreNativeManager.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreNativeManager.kt index 502870bd9f..663d0629d4 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreNativeManager.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreNativeManager.kt @@ -97,4 +97,4 @@ object CoreNativeManager { throw e } } -} +} \ No newline at end of file diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt index 1b8d6433cb..ca3100fed1 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt @@ -7,7 +7,7 @@ import com.v2ray.ang.dto.V2rayConfig import com.v2ray.ang.util.JsonUtil /** Combines v2rayNG-generated real-delay configurations into one probe core. */ -object ProbeConfigBuilder { +internal object ProbeConfigBuilder { data class Source(val guid: String, val config: V2rayConfig) fun build(sources: List, destination: String): ProbePlan { @@ -53,7 +53,7 @@ object ProbeConfigBuilder { ), ) return ProbePlan( - content = JsonUtil.toJsonPretty(config).orEmpty(), + content = JsonUtil.toJson(config), profiles = profiles, individualGuids = individualGuids, ) 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 bad83fe866..15ca659baa 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 @@ -159,7 +159,7 @@ class CoreTestService : Service() { } is RealPingEvent.Finish -> { - if (message.subscriptionId.isNotEmpty()) { + if (event.status == "0" && message.subscriptionId.isNotEmpty()) { if (MmkvManager.decodeSettingsBool(AppConfig.PREF_AUTO_REMOVE_INVALID_AFTER_TEST, false)) { AngConfigManager.removeInvalidServer(message.subscriptionId) } diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt index 8ee5eb9eed..d6f92b9922 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt @@ -72,7 +72,7 @@ class RealPingWorkerService( if (!finished) { LogUtil.e(AppConfig.TAG, "Probe batch failed", error) failPending() - finish("0") + finish("-1") } } } diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/service/SubscriptionUpdateService.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/service/SubscriptionUpdateService.kt index 88604bf2a4..36e91aef80 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/service/SubscriptionUpdateService.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/service/SubscriptionUpdateService.kt @@ -131,9 +131,9 @@ class SubscriptionUpdateService : Service() { } if (MmkvManager.decodeSettingsBool(AppConfig.PREF_AUTO_TEST_AFTER_UPDATE_SUBSCRIPTION, false)) { - testSubscriptionServers(sub) + val testCompleted = testSubscriptionServers(sub) - if (MmkvManager.decodeSettingsBool(AppConfig.PREF_AUTO_REMOVE_INVALID_AFTER_TEST, false)) { + if (testCompleted && MmkvManager.decodeSettingsBool(AppConfig.PREF_AUTO_REMOVE_INVALID_AFTER_TEST, false)) { LogUtil.i(AppConfig.TAG, "SubscriptionUpdateService: removing invalid servers for ${subItem.remarks}") showNotification( context = this, @@ -142,7 +142,7 @@ class SubscriptionUpdateService : Service() { ) AngConfigManager.removeInvalidServer(subId) } - if (MmkvManager.decodeSettingsBool(AppConfig.PREF_AUTO_SORT_AFTER_TEST, false)) { + if (testCompleted && MmkvManager.decodeSettingsBool(AppConfig.PREF_AUTO_SORT_AFTER_TEST, false)) { LogUtil.i(AppConfig.TAG, "SubscriptionUpdateService: sorting servers for ${subItem.remarks}") showNotification( context = this, @@ -156,7 +156,7 @@ class SubscriptionUpdateService : Service() { LogUtil.i(AppConfig.TAG, "SubscriptionUpdateService: Finished ${subItem.remarks}") } - private suspend fun testSubscriptionServers(sub: SubscriptionCache) { + private suspend fun testSubscriptionServers(sub: SubscriptionCache): Boolean { val subId = sub.guid LogUtil.i(AppConfig.TAG, "SubscriptionUpdateService: starting test phase for ${sub.subscription.remarks}") showNotification( @@ -166,27 +166,28 @@ class SubscriptionUpdateService : Service() { ) val guids = MmkvManager.decodeServerList(subId) - if (guids.isNotEmpty()) { - val deferred = CompletableDeferred() - lateinit var worker: RealPingWorkerService - worker = RealPingWorkerService( - context = this, - guids = guids, - onEvent = { event -> - handleWorkerEvent(event, sub.subscription.remarks) { - activeWorkers.remove(worker) - deferred.complete(Unit) - } + if (guids.isEmpty()) return true + + val deferred = CompletableDeferred() + lateinit var worker: RealPingWorkerService + worker = RealPingWorkerService( + context = this, + guids = guids, + onEvent = { event -> + handleWorkerEvent(event, sub.subscription.remarks) { completed -> + activeWorkers.remove(worker) + deferred.complete(completed) } - ) - activeWorkers.add(worker) - worker.start() - deferred.await() - LogUtil.i(AppConfig.TAG, "SubscriptionUpdateService: test phase finished for ${sub.subscription.remarks}") - } + }, + ) + activeWorkers.add(worker) + worker.start() + val completed = deferred.await() + LogUtil.i(AppConfig.TAG, "SubscriptionUpdateService: test phase finished for ${sub.subscription.remarks}") + return completed } - private fun handleWorkerEvent(event: RealPingEvent, remarks: String, onWorkerDone: () -> Unit) { + private fun handleWorkerEvent(event: RealPingEvent, remarks: String, onWorkerDone: (Boolean) -> Unit) { when (event) { is RealPingEvent.Progress -> { val notificationText = getString( @@ -207,7 +208,7 @@ class SubscriptionUpdateService : Service() { } is RealPingEvent.Finish -> { - onWorkerDone() + onWorkerDone(event.status == "0") } } } 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 30bcf98af3..44f21f2773 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 @@ -41,9 +41,7 @@ class MainRepository( private val _mainServiceEvent = MutableSharedFlow( replay = 0, - // A large policy group can finish an entire concurrency wave between - // main-thread dispatches. Keep every incremental result until the - // ViewModel coalesces it into one row update. + // Absorb large result bursts before the ViewModel coalesces UI updates. extraBufferCapacity = SERVICE_EVENT_BUFFER_CAPACITY, onBufferOverflow = BufferOverflow.DROP_OLDEST ) 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 cf589ec848..f68e9517e7 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 @@ -137,8 +137,9 @@ class MainViewModel( } is MainServiceEvent.MeasureConfigFinish -> { - testResultFlushJob?.cancel() + val scheduledFlush = testResultFlushJob testResultFlushJob = viewModelScope.launch { + scheduledFlush?.join() flushPendingTestResults() onTestsFinished() } @@ -150,8 +151,10 @@ class MainViewModel( pendingTestResults[result.guid] = result.delayMillis if (testResultFlushJob?.isActive == true) return testResultFlushJob = viewModelScope.launch { - delay(TEST_RESULT_FLUSH_INTERVAL_MS) - flushPendingTestResults() + while (pendingTestResults.isNotEmpty()) { + delay(TEST_RESULT_FLUSH_INTERVAL_MS) + flushPendingTestResults() + } } } @@ -826,6 +829,6 @@ class MainViewModel( } private companion object { - const val TEST_RESULT_FLUSH_INTERVAL_MS = 50L + const val TEST_RESULT_FLUSH_INTERVAL_MS = 500L } } From b91d5b02a415ac0e5675f84d83855bcdedc57ffb Mon Sep 17 00:00:00 2001 From: Eliot the Cougar Date: Fri, 14 Aug 2026 22:43:38 +0300 Subject: [PATCH 10/10] Stabilize observatory probe lifecycle --- .../com/v2ray/ang/core/CoreNativeManager.kt | 18 +-------------- .../ang/service/RealPingWorkerService.kt | 22 +++++++++---------- .../ang/service/SubscriptionUpdateService.kt | 6 ++++- .../com/v2ray/ang/ui/main/MainRepository.kt | 1 + .../com/v2ray/ang/ui/main/MainServiceEvent.kt | 3 ++- .../com/v2ray/ang/ui/main/MainViewModel.kt | 7 ++---- 6 files changed, 21 insertions(+), 36 deletions(-) diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreNativeManager.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreNativeManager.kt index 663d0629d4..124baa7091 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreNativeManager.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreNativeManager.kt @@ -67,22 +67,6 @@ object CoreNativeManager { } } - /** - * Measure outbound connection delay. - * - * @param config The configuration JSON string - * @param testUrl The URL to test against - * @return Delay in milliseconds, or -1 if test failed - */ - fun measureOutboundDelay(config: String, testUrl: String): Long { - return try { - Libv2ray.measureOutboundDelay(config, testUrl) - } catch (e: Exception) { - LogUtil.e(AppConfig.TAG, "Failed to measure outbound delay", e) - -1L - } - } - /** * Create a new core controller instance. * @@ -97,4 +81,4 @@ object CoreNativeManager { throw e } } -} \ No newline at end of file +} diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt index d6f92b9922..1c506dc0b6 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt @@ -4,7 +4,6 @@ import android.content.Context import android.os.SystemClock import com.v2ray.ang.AppConfig import com.v2ray.ang.core.CoreConfigManager -import com.v2ray.ang.core.CoreNativeManager import com.v2ray.ang.dto.ProbePlan import com.v2ray.ang.dto.RealPingEvent import com.v2ray.ang.enums.EConfigType @@ -108,17 +107,16 @@ class RealPingWorkerService( completeWork(guid, profileCompleted = true) } probeBatch(plan, concurrency) - probeIndividually(plan.individualGuids, concurrency) + probeIndividually(plan.individualGuids) } - private suspend fun probeIndividually(individualGuids: List, concurrency: Int) { - val dispatcher = Dispatchers.IO.limitedParallelism(concurrency) - individualGuids.map { guid -> - scope.launch(dispatcher) { - emitResult(guid, safelyProbe(guid, ::startRealPing)) - completeWork(guid, profileCompleted = true) - } - }.joinAll() + /** Each fallback needs its own Xray instance, so these cannot overlap safely. */ + private suspend fun probeIndividually(individualGuids: List) { + individualGuids.forEach { guid -> + currentCoroutineContext().ensureActive() + emitResult(guid, safelyProbe(guid, ::startRealPing)) + completeWork(guid, profileCompleted = true) + } } private suspend fun probeBatch(plan: ProbePlan, concurrency: Int) { @@ -160,7 +158,7 @@ class RealPingWorkerService( val activeGuids = retryGuids.filter(::isPending) if (activeGuids.isEmpty()) return if (activeGuids.size == 1) { - probeIndividually(activeGuids, concurrency) + probeIndividually(activeGuids) return } val halves = activeGuids.chunked((activeGuids.size + 1) / 2) @@ -251,7 +249,7 @@ class RealPingWorkerService( private fun startRealPing(guid: String): Long { val configResult = CoreConfigManager.getV2rayConfig4RealDelay(context, guid) if (!configResult.status) return -1L - return CoreNativeManager.measureOutboundDelay(configResult.content, SettingsManager.getDelayTestUrl()) + return controller.measureDelay(configResult.content, SettingsManager.getDelayTestUrl()) } private fun safelyProbe(guid: String, probe: (String) -> Long): Long = try { diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/service/SubscriptionUpdateService.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/service/SubscriptionUpdateService.kt index 36e91aef80..f8b6a47627 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/service/SubscriptionUpdateService.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/service/SubscriptionUpdateService.kt @@ -42,6 +42,8 @@ class SubscriptionUpdateService : Service() { private val activeWorkers = Collections.synchronizedList(mutableListOf()) private val updateSemaphore = Semaphore(2) + // Downloads may overlap, but native probe batches in this process may not. + private val probeSemaphore = Semaphore(1) override fun onCreate() { super.onCreate() @@ -131,7 +133,9 @@ class SubscriptionUpdateService : Service() { } if (MmkvManager.decodeSettingsBool(AppConfig.PREF_AUTO_TEST_AFTER_UPDATE_SUBSCRIPTION, false)) { - val testCompleted = testSubscriptionServers(sub) + val testCompleted = probeSemaphore.withPermit { + testSubscriptionServers(sub) + } if (testCompleted && MmkvManager.decodeSettingsBool(AppConfig.PREF_AUTO_REMOVE_INVALID_AFTER_TEST, false)) { LogUtil.i(AppConfig.TAG, "SubscriptionUpdateService: removing invalid servers for ${subItem.remarks}") 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 f70ba165c3..12ab44fa2c 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 @@ -8,6 +8,7 @@ import androidx.core.content.ContextCompat 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 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 002afc640b..506066c9bb 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,5 +1,6 @@ package com.v2ray.ang.ui.main +import com.v2ray.ang.dto.ConnectionTestResult import com.v2ray.ang.dto.RealPingResult sealed class MainServiceEvent { @@ -8,7 +9,7 @@ sealed class MainServiceEvent { data object StateStartSuccess : MainServiceEvent() data object StateStartFailure : MainServiceEvent() data object StateStopSuccess : MainServiceEvent() - data class MeasureDelaySuccess(val content: String) : MainServiceEvent() + data class MeasureDelayResult(val result: ConnectionTestResult) : MainServiceEvent() data class MeasureConfigSuccess(val result: RealPingResult) : MainServiceEvent() data class MeasureConfigNotify(val progress: String) : MainServiceEvent() data class MeasureConfigFinish(val finishedCount: 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 250258372b..cd6d757c06 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 @@ -154,10 +154,7 @@ class MainViewModel( if (delayMillis == null || delayMillis == server.testDelayMillis) { server } else { - server.copy( - testDelayMillis = delayMillis, - testDelayString = if (delayMillis == 0L) "" else "${delayMillis}ms", - ) + server.copy(testDelayMillis = delayMillis) } } } @@ -739,7 +736,7 @@ class MainViewModel( mutableServersForGroup(groupId).update { current -> current.map { server -> if (server.testDelayMillis == 0L) server - else server.copy(testDelayMillis = 0L, testDelayString = "") + else server.copy(testDelayMillis = 0L) } } testingGroupId = groupId