From 15cdd397cb95079d5dfca22277c38493f019f0c7 Mon Sep 17 00:00:00 2001 From: heaven Date: Tue, 16 Jun 2026 21:39:44 +0300 Subject: [PATCH] feat(service): add live throughput history ring to VpnState, fed each stats tick; cleared on connect/restart/stop --- .../vojo/proxy/service/ProxyVpnService.kt | 3 + .../java/chat/vojo/proxy/service/VpnState.kt | 93 +++++++++++++++++++ 2 files changed, 96 insertions(+) diff --git a/app/src/main/java/chat/vojo/proxy/service/ProxyVpnService.kt b/app/src/main/java/chat/vojo/proxy/service/ProxyVpnService.kt index e855872..90f0a9e 100644 --- a/app/src/main/java/chat/vojo/proxy/service/ProxyVpnService.kt +++ b/app/src/main/java/chat/vojo/proxy/service/ProxyVpnService.kt @@ -472,6 +472,9 @@ class ProxyVpnService : VpnService(), SocketProtector, DirectDialer { val s = Tun2Socks.nativeStats() if (s.size >= 4) { VpnState.setStats(VpnState.Stats(txBytes = s[1], rxBytes = s[3])) + // Same reading feeds the live throughput graph; rate/delta math + // (and the reset baseline) lives in VpnState so it stays atomic. + VpnState.pushRate(curTx = s[1], curRx = s[3], nowMs = System.currentTimeMillis()) } } // Refresh the notification with live traffic once a minute. diff --git a/app/src/main/java/chat/vojo/proxy/service/VpnState.kt b/app/src/main/java/chat/vojo/proxy/service/VpnState.kt index e95e21a..cc5d566 100644 --- a/app/src/main/java/chat/vojo/proxy/service/VpnState.kt +++ b/app/src/main/java/chat/vojo/proxy/service/VpnState.kt @@ -31,6 +31,26 @@ object VpnState { enum class Tone { NEUTRAL, OK, WARN, ERROR } } + /** + * Immutable ordered snapshot of recent throughput for the live graph — oldest at + * index 0, newest at [count]-1, rates in **bytes/sec**. [peak] is a decaying running + * max (recent activity, not a stuck all-session high) used for the graph's y-scale. + * [seq] increments on every emit so the UI can key its per-sample animation even + * after the ring saturates and [count] stops changing. + */ + class RateHistory( + val rx: IntArray, + val tx: IntArray, + val count: Int, + val peak: Int, + val seq: Int, + ) { + companion object { val EMPTY = RateHistory(IntArray(0), IntArray(0), 0, 0, 0) } + } + + /** Samples held in the throughput ring (CAP × stats cadence = visible window length). */ + const val RATE_CAP = 64 + private const val MAX_EVENTS = 30 private val eventSeq = AtomicLong(0) @@ -53,9 +73,31 @@ object VpnState { private val _events = MutableStateFlow>(emptyList()) val events: StateFlow> = _events.asStateFlow() + private val _rates = MutableStateFlow(RateHistory.EMPTY) + val rates: StateFlow = _rates.asStateFlow() + + // Ring backing + delta-baseline state for [pushRate]. [pushRate] runs on the stats + // coroutine while [clearRates] can run on the lifecycle thread during teardown, and + // Job.cancel() doesn't wait for an in-flight sample — so both guard their reads/writes + // of this shared state under [rateLock] (uncontended in steady state; one cheap lock + // per 2 s sample). The lock also gives the cross-thread happens-before the JMM needs. + private val rateLock = Any() + private val rxRing = IntArray(RATE_CAP) + private val txRing = IntArray(RATE_CAP) + private var ringCount = 0 + private var ringHead = 0 // index of the oldest live sample + private var decayPeak = 0 + private var lastTx = 0L + private var lastRx = 0L + private var lastSampleMs = 0L + private var rateSeq = 0 + internal fun setConnecting() { _error.value = null _status.value = Status.CONNECTING + // A fresh session (incl. a server-switch ACTION_RESTART, which routes through + // here) starts from an empty graph and a clean delta baseline. + clearRates() } internal fun setConnected(applied: Applied) { @@ -79,12 +121,63 @@ object VpnState { _connectedSince.value = 0L _stats.value = Stats() _applied.value = null + clearRates() } internal fun setStats(stats: Stats) { _stats.value = stats } + /** + * Feeds one cumulative-bytes reading; derives per-interval throughput and appends it + * to the ring. The first reading after a [clearRates] only establishes the delta + * baseline (emitting a rate here would divide the whole cumulative total by one + * interval — a phantom spike on connect/reconnect). + */ + internal fun pushRate(curTx: Long, curRx: Long, nowMs: Long) { + synchronized(rateLock) { + if (lastSampleMs == 0L) { + lastTx = curTx; lastRx = curRx; lastSampleMs = nowMs + return + } + val dt = (nowMs - lastSampleMs).coerceAtLeast(1) + // A negative delta = native counters reset (a new engine) or rolled over — + // treat that direction as idle this interval rather than as a huge spike. + val dTx = (curTx - lastTx).coerceAtLeast(0) + val dRx = (curRx - lastRx).coerceAtLeast(0) + val txRate = (dTx * 1000 / dt).coerceIn(0, Int.MAX_VALUE.toLong()).toInt() + val rxRate = (dRx * 1000 / dt).coerceIn(0, Int.MAX_VALUE.toLong()).toInt() + lastTx = curTx; lastRx = curRx; lastSampleMs = nowMs + + val idx = (ringHead + ringCount) % RATE_CAP + if (ringCount < RATE_CAP) { + rxRing[idx] = rxRate; txRing[idx] = txRate; ringCount++ + } else { + rxRing[ringHead] = rxRate; txRing[ringHead] = txRate + ringHead = (ringHead + 1) % RATE_CAP + } + decayPeak = maxOf((decayPeak * 0.94f).toInt(), rxRate, txRate) + + // Emit an immutable oldest→newest copy (≤ RATE_CAP ints ×2 once per interval). + val orx = IntArray(ringCount) + val otx = IntArray(ringCount) + for (i in 0 until ringCount) { + val r = (ringHead + i) % RATE_CAP + orx[i] = rxRing[r]; otx[i] = txRing[r] + } + _rates.value = RateHistory(orx, otx, ringCount, decayPeak, ++rateSeq) + } + } + + /** Drops all throughput history and the delta baseline; the graph falls back to idle. */ + internal fun clearRates() { + synchronized(rateLock) { + ringCount = 0; ringHead = 0; decayPeak = 0 + lastTx = 0L; lastRx = 0L; lastSampleMs = 0L; rateSeq = 0 + _rates.value = RateHistory.EMPTY + } + } + internal fun event(label: String, detail: String? = null, tone: TunnelEvent.Tone = TunnelEvent.Tone.NEUTRAL) { val ev = TunnelEvent(eventSeq.incrementAndGet(), System.currentTimeMillis(), label, detail, tone) _events.value = (_events.value + ev).takeLast(MAX_EVENTS)