feat(service): add live throughput history ring to VpnState, fed each stats tick; cleared on connect/restart/stop
This commit is contained in:
parent
68aa25f280
commit
15cdd397cb
2 changed files with 96 additions and 0 deletions
|
|
@ -472,6 +472,9 @@ class ProxyVpnService : VpnService(), SocketProtector, DirectDialer {
|
||||||
val s = Tun2Socks.nativeStats()
|
val s = Tun2Socks.nativeStats()
|
||||||
if (s.size >= 4) {
|
if (s.size >= 4) {
|
||||||
VpnState.setStats(VpnState.Stats(txBytes = s[1], rxBytes = s[3]))
|
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.
|
// Refresh the notification with live traffic once a minute.
|
||||||
|
|
|
||||||
|
|
@ -31,6 +31,26 @@ object VpnState {
|
||||||
enum class Tone { NEUTRAL, OK, WARN, ERROR }
|
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 const val MAX_EVENTS = 30
|
||||||
private val eventSeq = AtomicLong(0)
|
private val eventSeq = AtomicLong(0)
|
||||||
|
|
||||||
|
|
@ -53,9 +73,31 @@ object VpnState {
|
||||||
private val _events = MutableStateFlow<List<TunnelEvent>>(emptyList())
|
private val _events = MutableStateFlow<List<TunnelEvent>>(emptyList())
|
||||||
val events: StateFlow<List<TunnelEvent>> = _events.asStateFlow()
|
val events: StateFlow<List<TunnelEvent>> = _events.asStateFlow()
|
||||||
|
|
||||||
|
private val _rates = MutableStateFlow(RateHistory.EMPTY)
|
||||||
|
val rates: StateFlow<RateHistory> = _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() {
|
internal fun setConnecting() {
|
||||||
_error.value = null
|
_error.value = null
|
||||||
_status.value = Status.CONNECTING
|
_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) {
|
internal fun setConnected(applied: Applied) {
|
||||||
|
|
@ -79,12 +121,63 @@ object VpnState {
|
||||||
_connectedSince.value = 0L
|
_connectedSince.value = 0L
|
||||||
_stats.value = Stats()
|
_stats.value = Stats()
|
||||||
_applied.value = null
|
_applied.value = null
|
||||||
|
clearRates()
|
||||||
}
|
}
|
||||||
|
|
||||||
internal fun setStats(stats: Stats) {
|
internal fun setStats(stats: Stats) {
|
||||||
_stats.value = 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) {
|
internal fun event(label: String, detail: String? = null, tone: TunnelEvent.Tone = TunnelEvent.Tone.NEUTRAL) {
|
||||||
val ev = TunnelEvent(eventSeq.incrementAndGet(), System.currentTimeMillis(), label, detail, tone)
|
val ev = TunnelEvent(eventSeq.incrementAndGet(), System.currentTimeMillis(), label, detail, tone)
|
||||||
_events.value = (_events.value + ev).takeLast(MAX_EVENTS)
|
_events.value = (_events.value + ev).takeLast(MAX_EVENTS)
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue