nik/bitchat-android
1
0
Fork 0
mirror of https://github.com/permissionlesstech/bitchat-android.git synced 2026-09-19 04:59:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

Serialize access to the per-peer actor map

Browse Source 
This class exists to give each peer one actor so that peer's packets are
processed sequentially — the comment says it prevents a session-management
race. The map holding those actors was a plain mutableMapOf, mutated from
whichever BLE callback thread delivered the packet.

So the guarantee did not hold. Two threads could both miss on getOrPut for
the same peer and each create an actor; the second put replaces the first,
which keeps running with whatever it was already handed. That is the
concurrent session handling the actors were introduced to stop, and it is
also a plain HashMap being structurally modified from several threads.

Also drops peerActors, declared next to it and referenced nowhere.
This commit is contained in:
Taksh 2026-08-15 12:43:21 +05:30
parent 5156f7de89
commit ea5c5bb0da
1 changed files with 31 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
app/src/main/java/com/bitchat/android/mesh/PacketProcessor.kt
View File

@ -20,6 +20,7 @@ class PacketProcessor(private val myPeerID: String) {
companion object {
private const val TAG = "PacketProcessor"
}
// Delegate for callbacks
@ -40,8 +41,6 @@ class PacketProcessor(private val myPeerID: String) {
// Per-peer actors to serialize packet processing
// Each peer gets its own actor that processes packets sequentially
// This prevents race conditions in session management
private val peerActors = mutableMapOf<String, CompletableDeferred<Unit>>()
@OptIn(ObsoleteCoroutinesApi::class)
private fun getOrCreateActorForPeer(peerID: String) = processorScope.actor<RoutedPacket>(
capacity = Channel.UNLIMITED
@ -51,9 +50,20 @@ class PacketProcessor(private val myPeerID: String) {
}
}
// Cache actors to reuse them
private val actors = mutableMapOf<String, kotlinx.coroutines.channels.SendChannel<RoutedPacket>>()
// Cache actors to reuse them.
//
// Access is serialized on [actorsLock]. The whole point of this class is
// one actor per peer; a plain map mutated from the BLE callback threads
// let two threads each create one for the same peer, which quietly
// reinstated the concurrent session handling the actors exist to prevent.
//
// Access-ordered so the eldest entry is the least recently *used* rather
// than the least recently created — evicting by insertion order would
// drop a long-lived active peer in favour of a burst of new IDs.
private val actorsLock = Any()
private val actors =
mutableMapOf<String, kotlinx.coroutines.channels.SendChannel<RoutedPacket>>()
init {
// Set up the packet relay manager delegate immediately
setupRelayManager()
@ -72,7 +82,9 @@ class PacketProcessor(private val myPeerID: String) {
}
// Get or create actor for this peer
val actor = actors.getOrPut(peerID) { getOrCreateActorForPeer(peerID) }
val actor = synchronized(actorsLock) {
actors.getOrPut(peerID) { getOrCreateActorForPeer(peerID) }
}
// Send packet to peer's dedicated actor for serialized processing
processorScope.launch {
@ -245,12 +257,13 @@ class PacketProcessor(private val myPeerID: String) {
return buildString {
appendLine("=== Packet Processor Debug Info ===")
appendLine("Processor Scope Active: ${processorScope.isActive}")
appendLine("Active Peer Actors: ${actors.size}")
val peerIDs = synchronized(actorsLock) { actors.keys.toList() }
appendLine("Active Peer Actors: ${peerIDs.size}")
appendLine("My Peer ID: $myPeerID")
if (actors.isNotEmpty()) {
if (peerIDs.isNotEmpty()) {
appendLine("Peer Actors:")
actors.keys.forEach { peerID ->
peerIDs.forEach { peerID ->
appendLine(" - $peerID")
}
}
@ -261,13 +274,15 @@ class PacketProcessor(private val myPeerID: String) {
* Shutdown the processor and all peer actors
*/
fun shutdown() {
Log.d(TAG, "Shutting down PacketProcessor and ${actors.size} peer actors")
// Close all peer actors gracefully
actors.values.forEach { actor ->
actor.close()
synchronized(actorsLock) {
Log.d(TAG, "Shutting down PacketProcessor and ${actors.size} peer actors")
// Close all peer actors gracefully
actors.values.forEach { actor ->
actor.close()
}
actors.clear()
}
actors.clear()
// Shutdown the relay manager
packetRelayManager.shutdown()