From 1c9ae760d91af5656a92c9589cbc007a5817320b Mon Sep 17 00:00:00 2001 From: tapframe <85391825+tapframe@users.noreply.github.com> Date: Sat, 30 May 2026 03:08:26 +0530 Subject: [PATCH] Avoid blocking P2P teardown --- .../p2p/P2pStreamingEngine.android.kt | 108 +++++++++++++++--- 1 file changed, 90 insertions(+), 18 deletions(-) diff --git a/composeApp/src/androidMain/kotlin/com/nuvio/app/features/p2p/P2pStreamingEngine.android.kt b/composeApp/src/androidMain/kotlin/com/nuvio/app/features/p2p/P2pStreamingEngine.android.kt index 547dc4243..7497c0607 100644 --- a/composeApp/src/androidMain/kotlin/com/nuvio/app/features/p2p/P2pStreamingEngine.android.kt +++ b/composeApp/src/androidMain/kotlin/com/nuvio/app/features/p2p/P2pStreamingEngine.android.kt @@ -13,7 +13,6 @@ import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow import kotlinx.coroutines.isActive import kotlinx.coroutines.launch -import kotlinx.coroutines.runBlocking import kotlinx.coroutines.withContext import okhttp3.MediaType.Companion.toMediaType import okhttp3.OkHttpClient @@ -33,8 +32,11 @@ actual object P2pStreamingEngine { actual val state: StateFlow = _state.asStateFlow() private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO) + private val lifecycleLock = Any() private var statsJob: Job? = null + private var cleanupJob: Job? = null private var currentHash: String? = null + private var streamGeneration = 0L private var appContext: Context? = null private val binary = TorrServerBinary() private val api = TorrServerApi(binary) @@ -45,29 +47,37 @@ actual object P2pStreamingEngine { } actual suspend fun startStream(request: P2pStreamRequest): String = withContext(Dispatchers.IO) { - stopStream() + stopStreamNow(stopBinary = false) + val generation = nextStreamGeneration() _state.value = P2pStreamingState.Connecting try { binary.start() + ensureCurrentGeneration(generation) val magnetLink = buildMagnetUri(request.infoHash, request.trackers) Log.d(TAG, "Starting stream: $magnetLink") val hash = api.addTorrent(magnetLink) ?: throw P2pStreamingException("Failed to add torrent") - currentHash = hash + if (!attachTorrentIfCurrent(generation, hash)) { + api.dropTorrent(hash) + throw CancellationException("P2P stream start was cancelled") + } val resolvedIdx = resolveFileIndex( hash = hash, requestedIdx = request.fileIdx, filename = request.filename, ) + ensureCurrentGeneration(generation) + val streamUrl = api.getStreamUrl(magnetLink, resolvedIdx) Log.d(TAG, "Stream URL: $streamUrl") - startStatsPolling(hash) + startStatsPolling(hash, generation) + ensureCurrentGeneration(generation) _state.value = P2pStreamingState.Streaming( localUrl = streamUrl, downloadSpeed = 0, @@ -82,31 +92,88 @@ actual object P2pStreamingEngine { } catch (e: CancellationException) { throw e } catch (e: Exception) { - _state.value = P2pStreamingState.Error(e.message ?: "Unknown torrent error") + if (isCurrentGeneration(generation)) { + _state.value = P2pStreamingState.Error(e.message ?: "Unknown torrent error") + } throw e } } actual fun stopStream() { - statsJob?.cancel() - statsJob = null + scheduleStop(stopBinary = false) + } - currentHash?.let { hash -> + actual fun shutdown() { + scheduleStop(stopBinary = true) + } + + private fun scheduleStop(stopBinary: Boolean) { + val hash = detachActiveStream() + val previousCleanup = cleanupJob + cleanupJob = scope.launch { + previousCleanup?.join() + cleanupDetachedStream(hash, stopBinary) + } + } + + private suspend fun stopStreamNow(stopBinary: Boolean) { + cleanupJob?.join() + val hash = detachActiveStream() + cleanupDetachedStream(hash, stopBinary) + } + + private fun detachActiveStream(): String? { + val detached = synchronized(lifecycleLock) { + streamGeneration += 1 + val hash = currentHash + val job = statsJob + currentHash = null + statsJob = null + hash to job + } + detached.second?.cancel() + _state.value = P2pStreamingState.Idle + return detached.first + } + + private suspend fun cleanupDetachedStream(hash: String?, stopBinary: Boolean) { + hash?.let { try { - runBlocking(Dispatchers.IO) { - api.dropTorrent(hash) - } + api.dropTorrent(it) } catch (e: Exception) { Log.w(TAG, "Error dropping torrent", e) } } - currentHash = null - _state.value = P2pStreamingState.Idle + + if (stopBinary) { + try { + binary.stop() + } catch (e: Exception) { + Log.w(TAG, "Error stopping TorrServer", e) + } + } } - actual fun shutdown() { - stopStream() - binary.stop() + private fun nextStreamGeneration(): Long = + synchronized(lifecycleLock) { + streamGeneration += 1 + streamGeneration + } + + private fun attachTorrentIfCurrent(generation: Long, hash: String): Boolean = + synchronized(lifecycleLock) { + if (streamGeneration != generation) return@synchronized false + currentHash = hash + true + } + + private fun isCurrentGeneration(generation: Long): Boolean = + synchronized(lifecycleLock) { streamGeneration == generation } + + private fun ensureCurrentGeneration(generation: Long) { + if (!isCurrentGeneration(generation)) { + throw CancellationException("P2P stream start was cancelled") + } } private fun buildMagnetUri(infoHash: String, extraTrackers: List): String { @@ -176,14 +243,19 @@ actual object P2pStreamingEngine { return result } - private fun startStatsPolling(hash: String) { + private fun startStatsPolling(hash: String, generation: Long) { statsJob?.cancel() statsJob = scope.launch { while (isActive) { + if (!isCurrentGeneration(generation)) return@launch try { val stats = api.getTorrentStats(hash) val currentState = _state.value - if (stats != null && currentState is P2pStreamingState.Streaming) { + if ( + stats != null && + currentState is P2pStreamingState.Streaming && + isCurrentGeneration(generation) + ) { _state.value = currentState.copy( downloadSpeed = stats.downloadSpeed, uploadSpeed = stats.uploadSpeed,