diff --git a/app/src/main/AndroidManifest.xml b/app/src/main/AndroidManifest.xml index 1e167ba..bdb6768 100644 --- a/app/src/main/AndroidManifest.xml +++ b/app/src/main/AndroidManifest.xml @@ -6,6 +6,11 @@ + + diff --git a/app/src/main/java/org/prolibertate/games/net/LanTransport.kt b/app/src/main/java/org/prolibertate/games/net/LanTransport.kt index 506b550..160ff4a 100644 --- a/app/src/main/java/org/prolibertate/games/net/LanTransport.kt +++ b/app/src/main/java/org/prolibertate/games/net/LanTransport.kt @@ -3,6 +3,7 @@ package org.prolibertate.games.net import android.content.Context import android.net.nsd.NsdManager import android.net.nsd.NsdServiceInfo +import android.net.wifi.WifiManager import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.channels.awaitClose @@ -31,17 +32,49 @@ class LanTransport(private val context: Context) : Transport { private val nsdManager: NsdManager get() = context.getSystemService(Context.NSD_SERVICE) as NsdManager + private val wifiManager: WifiManager + get() = context.applicationContext.getSystemService(Context.WIFI_SERVICE) as WifiManager + private var serverSocket: ServerSocket? = null private var registrationListener: NsdManager.RegistrationListener? = null private var discoveryListener: NsdManager.DiscoveryListener? = null + private var wifiLock: WifiManager.WifiLock? = null override fun isAvailable(): Boolean = true + /** + * Keeps the Wi-Fi radio properly awake for as long as there is a game on. + * + * Android puts the radio into power save when the screen goes off, which is + * what a locked phone is, and a link left to it drops out from under a game + * that is still being played. FULL_HIGH_PERF is the mode that says otherwise + * — the deprecation notice points at LOW_LATENCY, which is the wrong one + * here: that one only applies while the screen is on and the app is in + * front, which is precisely the case that was never broken. + */ + @Suppress("DEPRECATION") + private fun holdRadio() { + if (wifiLock?.isHeld == true) return + val lock = wifiLock ?: runCatching { + wifiManager.createWifiLock(WifiManager.WIFI_MODE_FULL_HIGH_PERF, WIFI_LOCK_TAG) + .also { it.setReferenceCounted(false) } + }.getOrNull() ?: return + wifiLock = lock + runCatching { lock.acquire() } + } + + private fun releaseRadio() { + wifiLock?.let { lock -> runCatching { if (lock.isHeld) lock.release() } } + wifiLock = null + } + // ----------------------------------------------------------------------- // Hosting // ----------------------------------------------------------------------- override fun host(displayName: String, scope: CoroutineScope): Flow = callbackFlow { + holdRadio() + // Port 0 lets the OS pick a free port, which is then advertised. val server = ServerSocket(0) serverSocket = server @@ -66,6 +99,7 @@ class LanTransport(private val context: Context) : Transport { while (isActive && !server.isClosed) { val socket = runCatching { server.accept() }.getOrNull() ?: break socket.tcpNoDelay = true + socket.keepAlive = true val connection = StreamConnection( peerId = socket.inetAddress?.hostAddress ?: "unknown", kind = TransportKind.LAN, @@ -90,6 +124,8 @@ class LanTransport(private val context: Context) : Transport { // ----------------------------------------------------------------------- override fun discover(scope: CoroutineScope): Flow> = callbackFlow { + holdRadio() + val found = Collections.synchronizedMap(linkedMapOf()) fun publish() { @@ -177,9 +213,11 @@ class LanTransport(private val context: Context) : Transport { override suspend fun join(host: DiscoveredHost, scope: CoroutineScope): Connection = withContext(Dispatchers.IO) { + holdRadio() val socket = Socket() socket.connect(InetSocketAddress(host.address, host.port), CONNECT_TIMEOUT_MS) socket.tcpNoDelay = true + socket.keepAlive = true StreamConnection( peerId = host.id, kind = TransportKind.LAN, @@ -207,9 +245,11 @@ class LanTransport(private val context: Context) : Transport { stopDiscovery() runCatching { serverSocket?.close() } serverSocket = null + releaseRadio() } private companion object { const val CONNECT_TIMEOUT_MS = 8_000 + const val WIFI_LOCK_TAG = "ProLibertateGames:lan" } } diff --git a/app/src/main/java/org/prolibertate/games/net/LobbyController.kt b/app/src/main/java/org/prolibertate/games/net/LobbyController.kt index 22cefd2..f4af96d 100644 --- a/app/src/main/java/org/prolibertate/games/net/LobbyController.kt +++ b/app/src/main/java/org/prolibertate/games/net/LobbyController.kt @@ -3,6 +3,7 @@ package org.prolibertate.games.net import android.content.Context import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job +import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow @@ -10,6 +11,7 @@ import kotlinx.coroutines.launch import org.prolibertate.games.game.engine.PlayerKind import org.prolibertate.games.game.engine.PlayerSlot import org.prolibertate.games.game.engine.TableConfig +import java.util.concurrent.atomic.AtomicBoolean /** * The pre-game lobby. @@ -48,8 +50,27 @@ class LobbyController( /** Peer id -> live connection, for every seated remote player. */ private val connections = linkedMapOf() private var clientLink: Connection? = null + private var clientJob: Job? = null private val jobs = mutableListOf() + /** + * Where this device joined, so it can find its way back. + * + * The host goes on listening on the same port for as long as it is hosting, + * which is the whole match, so a guest that has lost its link can dial the + * address it already knows rather than go round discovery again. + */ + private var joinedHost: DiscoveredHost? = null + private var joinedName: String = "" + private var joinedPeerId: String = "" + private val reconnecting = AtomicBoolean(false) + + /** + * The table currently being played, if any, so a replaced link can be given + * to it. Set by the match screen for as long as it is on screen. + */ + var rebinder: LinkRebinder? = null + val hostConnections: List get() = connections.values.toList() val clientConnection: Connection? get() = clientLink @@ -118,6 +139,34 @@ class LobbyController( private suspend fun seatPlayer(connection: Connection, hello: Hello) { val current = _state.value val seats = current.seats.toMutableList() + + // A device the table already knows is not somebody new asking for a + // seat: it is a player whose link went away — a phone locked long + // enough for the radio to drop it is the usual way — coming back to the + // one it already has. Seating it again would find no free seat and turn + // it away from its own game, so it is put back where it was. + val returning = seats.indexOfFirst { it.deviceId == hello.peerId } + if (returning >= 0) { + val stale = seats[returning].peerId?.let { connections.remove(it) } + seats[returning] = seats[returning].copy( + name = hello.displayName, + kind = PlayerKind.HUMAN_REMOTE, + peerId = connection.peerId, + ) + connections[connection.peerId] = connection + if (stale != null && stale !== connection) runCatching { stale.close() } + _state.value = current.copy( + seats = seats, + message = "${hello.displayName} is back.", + ) + connection.send(Welcome(current.hostName, current.gameId.orEmpty(), accepted = true)) + broadcastLobby() + // A match already under way keeps the seat and the position it is + // sitting in. Only the link is new. + rebinder?.rebindPeer(seats[returning].seat, connection) + return + } + val index = seats.indexOfFirst { it.kind == PlayerKind.AI } if (index < 0) { connection.send( @@ -208,45 +257,122 @@ class LobbyController( fun join(host: DiscoveredHost, displayName: String, peerId: String) { jobs += scope.launch { _state.value = _state.value.copy(message = "Connecting to ${host.name}…") - val transport = transports.firstOrNull { it.kind == host.kind } ?: return@launch - val connection = runCatching { transport.join(host, scope) }.getOrElse { error -> - _state.value = _state.value.copy(message = "Couldn't connect: ${error.message}") - return@launch - } - clientLink = connection + joinedHost = host + joinedName = displayName + joinedPeerId = peerId + val connection = dial(host) ?: return@launch + attachClientLink(connection) + } + } - launch { - connection.incoming.collect { message -> - when (message) { - is Welcome -> _state.value = if (message.accepted) { - _state.value.copy( - connected = true, - gameId = message.gameId, - hostName = message.hostName, - message = "Joined ${message.hostName}. Waiting for the host to start…", - ) - } else { - _state.value.copy(message = message.reason ?: "The host declined.") - } + private suspend fun dial(host: DiscoveredHost): Connection? { + val transport = transports.firstOrNull { it.kind == host.kind } ?: return null + return runCatching { transport.join(host, scope) }.getOrElse { error -> + _state.value = _state.value.copy(message = "Couldn't connect: ${error.message}") + null + } + } - is LobbyUpdate -> _state.value = _state.value.copy( - seats = message.seats, + /** + * Starts reading [connection] as this device's way to the host, and + * introduces itself on it. + * + * Any previous reader is stopped first: collecting a connection's messages + * never ends on its own, so a link that has been replaced would otherwise + * leave its collector behind writing into the same lobby state. + */ + private suspend fun attachClientLink(connection: Connection) { + clientJob?.cancel() + clientLink = connection + clientJob = scope.launch { + connection.incoming.collect { message -> + when (message) { + is Welcome -> _state.value = if (message.accepted) { + _state.value.copy( + connected = true, gameId = message.gameId, - optionsJson = message.optionsJson, hostName = message.hostName, + message = "Joined ${message.hostName}. Waiting for the host to start…", ) - - is StartGame -> _state.value = _state.value.copy(started = message.config) - - is Bye -> _state.value = - _state.value.copy(connected = false, message = "The host left.") - - else -> Unit + } else { + _state.value.copy(message = message.reason ?: "The host declined.") } + + is LobbyUpdate -> _state.value = _state.value.copy( + seats = message.seats, + gameId = message.gameId, + optionsJson = message.optionsJson, + hostName = message.hostName, + ) + + is StartGame -> _state.value = _state.value.copy(started = message.config) + + is Bye -> { + _state.value = + _state.value.copy(connected = false, message = "The host left.") + // A link that has gone is worth replacing while there is + // still a table to come back to. In its own job, because + // taking the new link into use stops this one. + if (rebinder != null) jobs += scope.launch { reconnect() } + } + + else -> Unit } } + } - connection.send(Hello(peerId = peerId, displayName = displayName)) + connection.send(Hello(peerId = joinedPeerId, displayName = joinedName)) + } + + /** + * Client: makes sure there is still a way to the host, and asks for the + * table again. + * + * Called when the app comes back to the front. A phone that was locked when + * its turn came round has missed whatever the host pushed in the meantime — + * a client draws nothing but what it was sent — and may have lost the link + * entirely while the radio was down, in which case there is no point asking + * politely and it dials again instead. + * + * Costs nothing when nothing is wrong: a live link is sent one [Resync], + * and the answer is the state it already has. + */ + fun restoreLink() { + if (joinedHost == null) return + jobs += scope.launch { + val live = clientLink + if (live != null && live.isOpen) live.send(Resync) else reconnect() + } + } + + /** + * Dials the host again and hands the new link to whatever is being played. + * + * One attempt at a time: the link can be found to be gone twice over — the + * reader announcing it and the screen coming back both arrive at the same + * conclusion — and two dials would leave the host holding two links for one + * seat, having closed the one actually in use. + */ + private suspend fun reconnect() { + val host = joinedHost ?: return + if (!reconnecting.compareAndSet(false, true)) return + try { + repeat(RECONNECT_ATTEMPTS) { attempt -> + if (attempt > 0) delay(RECONNECT_BACKOFF_MILLIS * attempt) + if (clientLink?.isOpen == true) return + val connection = dial(host) + if (connection != null) { + // Hello first and rebind second, in that order: the host has + // to have put this device back in its seat before it can + // answer the table's resync with anything. + attachClientLink(connection) + rebinder?.rebindHostLink(connection) + return + } + } + _state.value = _state.value.copy(message = "Couldn't get back to ${host.name}.") + } finally { + reconnecting.set(false) } } @@ -257,11 +383,28 @@ class LobbyController( fun stop() { jobs.forEach { it.cancel() } jobs.clear() + clientJob?.cancel() + clientJob = null connections.values.forEach { runCatching { it.close() } } connections.clear() runCatching { clientLink?.close() } clientLink = null + // Nothing left to get back to, and nothing left to hand a link to. + joinedHost = null + joinedName = "" + joinedPeerId = "" + rebinder = null transports.forEach { runCatching { it.stop() } } _state.value = State() } + + private companion object { + /** + * Enough to ride out a radio that has not finished waking up, and few + * enough that a host which is genuinely gone is reported as gone rather + * than dialled at forever. + */ + const val RECONNECT_ATTEMPTS = 4 + const val RECONNECT_BACKOFF_MILLIS = 750L + } } diff --git a/app/src/main/java/org/prolibertate/games/net/MatchController.kt b/app/src/main/java/org/prolibertate/games/net/MatchController.kt index 89dde9c..d175acd 100644 --- a/app/src/main/java/org/prolibertate/games/net/MatchController.kt +++ b/app/src/main/java/org/prolibertate/games/net/MatchController.kt @@ -2,6 +2,7 @@ package org.prolibertate.games.net import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.Job import kotlinx.coroutines.delay import kotlinx.coroutines.withContext import kotlinx.coroutines.flow.MutableStateFlow @@ -56,7 +57,7 @@ class MatchController( * waits for [confirmAdvance]. */ private val awaitsConfirmation: (S) -> Boolean = { false }, -) { +) : LinkRebinder { enum class Role { HOST, CLIENT } @@ -111,15 +112,32 @@ class MatchController( private var released = false private val peers = mutableMapOf() + + /** + * The reader attached to each peer's link. + * + * Held so a replaced link's collector can be stopped. Collecting a + * [Connection.incoming] never ends on its own — a SharedFlow has no + * completion — so a link swapped out mid-match would otherwise leave its + * collector behind for the rest of the game. + */ + private val peerJobs = mutableMapOf() private var hostLink: Connection? = null + private var hostLinkJob: Job? = null private val seatKinds: Map = config.seats.associate { it.seat to it.kind } - private val seatPeers: Map = - config.seats.associate { it.seat to it.peerId } + + /** + * Which link each seat is reached on. Mutable because a player who drops + * and comes back is the same player in the same seat, but not necessarily + * on the same address — a reconnecting phone can be handed a different one. + */ + private val seatPeers: MutableMap = + config.seats.associate { it.seat to it.peerId }.toMutableMap() private fun nameOfPeer(peerId: String): String = - config.seats.firstOrNull { it.peerId == peerId }?.name ?: "The other player" + config.seats.firstOrNull { it.peerId == peerId }?.name ?: "the other player" /** * Records that this table cannot be played any further. @@ -140,12 +158,7 @@ class MatchController( /** Host: begin the match and start pushing state to [clients]. */ fun startAsHost(clients: List) { check(role == Role.HOST) { "Not hosting" } - clients.forEach { connection -> - peers[connection.peerId] = connection - scope.launch { - connection.incoming.collect { message -> onHostMessage(connection, message) } - } - } + clients.forEach { connection -> listenToPeer(connection) } scope.launch { val initial = runCatching { rules.initialState(config) }.getOrElse { error -> // A table the rules will not accept is a dead end, not a reason @@ -162,11 +175,18 @@ class MatchController( } } + private fun listenToPeer(connection: Connection) { + peers[connection.peerId] = connection + peerJobs[connection.peerId] = scope.launch { + connection.incoming.collect { message -> onHostMessage(connection, message) } + } + } + /** Client: render whatever arrives from [link]. */ fun startAsClient(link: Connection) { check(role == Role.CLIENT) { "Not a client" } hostLink = link - scope.launch { + hostLinkJob = scope.launch { link.incoming // Asked for from inside onSubscription, which is what makes it // reliable: the request cannot leave before this collector is @@ -196,11 +216,14 @@ class MatchController( is Rejected -> _notice.value = message.reason is Bye -> { - // The host has gone: the game cannot go on, and - // saying so is what gives the screen something to - // offer a way out of. + // The link to the host has gone, either because the + // host left or because the link died under it. The + // two are indistinguishable from this end, so this + // says only what is actually known — and it stops + // the table, which gives the screen a way out to + // offer if getting back in does not work. _legalMoves.value = emptyList() - abandon("The host left the game.") + abandon("The connection to the host was lost.") } else -> Unit @@ -209,6 +232,59 @@ class MatchController( } } + // ----------------------------------------------------------------------- + // Getting back in + // ----------------------------------------------------------------------- + + /** + * Host: a player who dropped is back, on [connection]. + * + * The seat, and the position it is sitting in, were never given up — only + * the link was — so nothing here touches the game. The old link is let go + * of, the new one is read from, and the seat is sent the table as it now + * stands so the returning player has something to play on. + */ + override fun rebindPeer(seat: Int, connection: Connection) { + if (role != Role.HOST) return + val previous = seatPeers[seat] + if (previous != null) { + peerJobs.remove(previous)?.cancel() + // Not closed: the lobby owns these sockets and has already dealt + // with the one being replaced. Closing a link this end has stopped + // reading would only race the reader that is on its way out. + peers.remove(previous) + } + seatPeers[seat] = connection.peerId + listenToPeer(connection) + // Only once everybody is reachable again. At a table of four, one player + // coming back says nothing about the other one who has not. + if (seatPeers.values.filterNotNull().all { peers[it]?.isOpen == true }) { + clearAbandoned() + } + scope.launch { sendStateTo(connection) } + } + + /** + * Client: the way back to the host is [connection]. + * + * The collector asks for a resync as it attaches, which is what fills the + * screen back in: everything a client draws came off the wire, and whatever + * was pushed while the link was down was pushed into nothing. + */ + override fun rebindHostLink(connection: Connection) { + if (role != Role.CLIENT) return + if (hostLink === connection) return + hostLinkJob?.cancel() + clearAbandoned() + startAsClient(connection) + } + + /** Puts a table that was stopped back into play. */ + private fun clearAbandoned() { + _abandoned.value = null + _notice.value = null + } + // ----------------------------------------------------------------------- // Input // ----------------------------------------------------------------------- @@ -232,10 +308,15 @@ class MatchController( private suspend fun onHostMessage(connection: Connection, message: NetMessage) { if (message is Bye) { - // One of the players has left. The host keeps the position — there is - // nothing wrong with it — but a table that cannot go on should say so - // rather than leave somebody waiting on a move that is not coming. - abandon("${nameOfPeer(connection.peerId)} left the game.") + // Anything arriving on a link that has since been replaced belongs + // to the link, not to the seat: a player who dropped and came back + // must not be reported as having left on the way in. + if (peers[connection.peerId] !== connection) return + // One of the players is no longer reachable. The host keeps the + // position — there is nothing wrong with it — but a table that + // cannot go on should say so rather than leave somebody waiting on a + // move that is not coming. + abandon("The connection to ${nameOfPeer(connection.peerId)} was lost.") return } if (message is Resync) { @@ -433,8 +514,12 @@ class MatchController( * [StreamConnection.close], which this leans on to not block. */ fun close() { + peerJobs.values.forEach { it.cancel() } + peerJobs.clear() peers.values.forEach { runCatching { it.close() } } peers.clear() + hostLinkJob?.cancel() + hostLinkJob = null runCatching { hostLink?.close() } hostLink = null } diff --git a/app/src/main/java/org/prolibertate/games/net/Protocol.kt b/app/src/main/java/org/prolibertate/games/net/Protocol.kt index 789bb7f..109b314 100644 --- a/app/src/main/java/org/prolibertate/games/net/Protocol.kt +++ b/app/src/main/java/org/prolibertate/games/net/Protocol.kt @@ -16,7 +16,14 @@ import org.prolibertate.games.game.engine.TableConfig * takes effect. Each client is sent a state redacted for its own seat, so an * opponent's hand is never transmitted in the first place. */ -const val PROTOCOL_VERSION = 1 +/** + * 2 added the heartbeat. A peer that does not answer [Ping] cannot be told + * apart from one that has gone away, so both ends have to speak it or a quiet + * link would be dropped as dead the first time nobody moved for a while. The + * lobby already refuses a mismatch with "update both devices", which is the + * honest thing to say about it. + */ +const val PROTOCOL_VERSION = 2 /** Service type and name advertised over mDNS. */ const val SERVICE_TYPE = "_plgames._tcp" @@ -116,6 +123,30 @@ data object Resync : NetMessage @SerialName("rejected") data class Rejected(val reason: String) : NetMessage +/** + * Sent to a link nobody has spoken on for a while, and answered with [Pong]. + * + * Two things need it. A phone whose screen has gone off puts its Wi-Fi radio + * into power save and, given long enough, stops the app's networking + * altogether; an access point ages an idle association out of its tables. A + * link carrying nothing between one player's turn and the next is idle for + * minutes at a time, which is exactly what both of those collect. And a TCP + * link that dies that way dies quietly: the read never returns and the next + * write is buffered into a socket that no longer goes anywhere, so without + * traffic of its own the app cannot tell a link that is waiting from one that + * is gone. + * + * Neither reaches a screen — see [StreamConnection], which answers and swallows + * them, so nothing above the transport has to know they exist. + */ +@Serializable +@SerialName("ping") +data object Ping : NetMessage + +@Serializable +@SerialName("pong") +data object Pong : NetMessage + @Serializable @SerialName("bye") data object Bye : NetMessage diff --git a/app/src/main/java/org/prolibertate/games/net/Transport.kt b/app/src/main/java/org/prolibertate/games/net/Transport.kt index 8e9a6b0..8adc43e 100644 --- a/app/src/main/java/org/prolibertate/games/net/Transport.kt +++ b/app/src/main/java/org/prolibertate/games/net/Transport.kt @@ -3,11 +3,14 @@ package org.prolibertate.games.net import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.NonCancellable +import kotlinx.coroutines.delay import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.SharedFlow import kotlinx.coroutines.isActive import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock import kotlinx.coroutines.withContext import kotlinx.serialization.encodeToString import java.io.BufferedReader @@ -60,12 +63,35 @@ class StreamConnection( private val output: OutputStream, private val scope: CoroutineScope, private val onClosed: (StreamConnection) -> Unit = {}, + /** How often a link with nothing to say is pinged. */ + private val heartbeatMillis: Long = HEARTBEAT_MILLIS, + /** How long a link may go completely silent before it counts as gone. */ + private val silenceMillis: Long = SILENCE_MILLIS, ) : Connection { private val reader: BufferedReader = input.bufferedReader() private val writer: BufferedWriter = output.bufferedWriter() + private val writeLock = Mutex() private val closed = AtomicBoolean(false) + /** + * Whether this end closed the link on purpose, as leaving a game does. + * + * Kept apart from [closed] because the two answer different questions. A + * link is closed either way; only one of the two ways is worth telling + * anybody about, and the heartbeat closes links that very much are. + */ + private val closedOnPurpose = AtomicBoolean(false) + + /** + * When the peer was last heard from, on the monotonic clock. + * + * Wall time would do the wrong thing across a clock adjustment, and + * SystemClock is Android's, which the unit tests do not have. + */ + @Volatile + private var lastHeardNanos: Long = System.nanoTime() + private val _incoming = MutableSharedFlow(replay = 0, extraBufferCapacity = 64) override val incoming: SharedFlow = _incoming @@ -73,8 +99,6 @@ class StreamConnection( init { scope.launch(Dispatchers.IO) { - // Told apart from a link this side closed on purpose, because only - // one of the two is worth telling anybody about. var droppedByPeer = false try { while (isActive && !closed.get()) { @@ -83,11 +107,22 @@ class StreamConnection( droppedByPeer = true break } + // Anything at all counts as a sign of life, including a line + // this version cannot make sense of: the point is that the + // peer is still there and still talking. + lastHeardNanos = System.nanoTime() if (line.isBlank()) continue val message = runCatching { protocolJson.decodeFromString(line) }.getOrNull() ?: continue // ignore anything we can't parse - _incoming.emit(message) + // The heartbeat is the transport's own business and stops + // here. Answering on this thread is safe: send() hops to IO, + // which is where this already is. + when (message) { + is Ping -> send(Pong) + is Pong -> Unit + else -> _incoming.emit(message) + } } } catch (_: Exception) { // A dropped link is normal: someone walked out of range or @@ -98,30 +133,62 @@ class StreamConnection( // lock, so it is the only one that can close it without // waiting. See close(). runCatching { reader.close() } - val weClosedIt = closed.get() - close() + val deliberate = closedOnPurpose.get() + shutdown() // A link that died on its own reaches the screens on the same - // channel as everything else, so leaving a game is something the + // channel as everything else, so losing a player is something the // other end finds out about the same way it finds out about a // move. Withdrawing deliberately needs no announcement, and the // emit is not cancellable — the whole point of it is to arrive // while everything around it is being torn down. - if (droppedByPeer && !weClosedIt) { + if (droppedByPeer && !deliberate) { withContext(NonCancellable) { runCatching { _incoming.emit(Bye) } } } } } + + // Keeps the link warm and, more to the point, keeps it answerable: a + // link nobody has spoken on cannot be told from a link that no longer + // exists, and between two turns of a card game that is most of the game. + scope.launch(Dispatchers.IO) { + while (isActive && !closed.get()) { + delay(heartbeatMillis) + if (closed.get()) break + val silentFor = (System.nanoTime() - lastHeardNanos) / 1_000_000 + if (silentFor > silenceMillis) { + // Closing the stream is what unblocks the read; the reader + // then announces the drop from its own thread, exactly as it + // would for a peer that had hung up. + shutdown() + break + } + send(Ping) + } + } } override suspend fun send(message: NetMessage) { if (closed.get()) return withContext(Dispatchers.IO) { - try { - writer.write(protocolJson.encodeToString(message)) - writer.newLine() - writer.flush() - } catch (_: Exception) { - close() + // One message at a time. The framing is what makes a line a message, + // and a write is three calls — the text, the newline, the flush — so + // two of them running at once would splice one message through the + // middle of another and neither would arrive. The heartbeat is what + // makes that likely: it writes on its own schedule, without caring + // what the game happens to be sending. + writeLock.withLock { + try { + writer.write(protocolJson.encodeToString(message)) + writer.newLine() + writer.flush() + } catch (_: Exception) { + // Not close(): a write that failed is a link that broke, and + // the screen has to hear about that. Going through close() + // here used to mark it deliberate, so a player whose link + // had died tapped a board that answered nothing and was + // never told why. + shutdown() + } } } } @@ -143,11 +210,52 @@ class StreamConnection( * buffered in it. */ override fun close() { + closedOnPurpose.set(true) + shutdown() + } + + /** Lets go of the streams without saying anything about why. */ + private fun shutdown() { if (!closed.compareAndSet(false, true)) return runCatching { input.close() } runCatching { output.close() } onClosed(this) } + + private companion object { + /** + * Often enough to hold an access point's idea of the association open + * and to notice a dead link inside a turn, seldom enough that it is a + * few dozen bytes a minute. + */ + const val HEARTBEAT_MILLIS = 5_000L + + /** + * Several missed beats rather than one. A phone with its screen off + * answers late — the radio is in power save and the timer that would + * have sent the ping is deferred — and a link is not dead just because + * it is slow. + */ + const val SILENCE_MILLIS = 25_000L + } +} + +/** + * A table that can be handed a replacement for a link it was given. + * + * A dropped link is not the end of a game — the position is untouched and the + * player is still sitting there — so the lobby, which owns the sockets and goes + * on listening for the whole match, reconnects and passes the new link through + * this. Narrow on purpose: the lobby has no business knowing what game is being + * played on the other side of it. + */ +interface LinkRebinder { + + /** Host: [connection] is the seat's link from now on. */ + fun rebindPeer(seat: Int, connection: Connection) + + /** Client: [connection] is the way to the host from now on. */ + fun rebindHostLink(connection: Connection) } /** diff --git a/app/src/main/java/org/prolibertate/games/ui/PlayScreen.kt b/app/src/main/java/org/prolibertate/games/ui/PlayScreen.kt index 3359b69..4235d36 100644 --- a/app/src/main/java/org/prolibertate/games/ui/PlayScreen.kt +++ b/app/src/main/java/org/prolibertate/games/ui/PlayScreen.kt @@ -9,6 +9,9 @@ import androidx.compose.runtime.mutableIntStateOf import androidx.compose.runtime.remember import androidx.compose.runtime.setValue import androidx.compose.runtime.rememberCoroutineScope +import androidx.compose.ui.platform.LocalLifecycleOwner +import androidx.lifecycle.Lifecycle +import androidx.lifecycle.LifecycleEventObserver import org.prolibertate.games.game.GameCatalog import org.prolibertate.games.game.chess.ChessAi import org.prolibertate.games.game.chess.ChessMove @@ -683,6 +686,32 @@ private fun StartMatch( } } DisposableEffect(controller) { - onDispose { controller.close() } + // The lobby owns the sockets and goes on listening for the whole match, + // so it is the lobby that reconnects a dropped player — and this is + // where it finds the table to hand the new link to. + env.lobby.rebinder = controller + onDispose { + env.lobby.rebinder = null + controller.close() + } + } + + // Coming back to a table that has been out of sight. + // + // A locked phone is a stopped activity, and its Wi-Fi radio goes to sleep + // with the screen. Anything the host pushed in the meantime — the player's + // own turn arriving is the one that matters — was pushed at a link that may + // no longer have been there, and a client draws nothing but what it was + // sent. So the table is asked for again on the way back in, over a fresh + // link if the old one did not survive. + val lifecycleOwner = LocalLifecycleOwner.current + DisposableEffect(lifecycleOwner, route.hosting) { + val observer = LifecycleEventObserver { _, event -> + if (event == Lifecycle.Event.ON_START && !route.hosting) { + env.lobby.restoreLink() + } + } + lifecycleOwner.lifecycle.addObserver(observer) + onDispose { lifecycleOwner.lifecycle.removeObserver(observer) } } } diff --git a/app/src/test/java/org/prolibertate/games/net/MatchControllerNetworkTest.kt b/app/src/test/java/org/prolibertate/games/net/MatchControllerNetworkTest.kt index 8b37c9b..953aebb 100644 --- a/app/src/test/java/org/prolibertate/games/net/MatchControllerNetworkTest.kt +++ b/app/src/test/java/org/prolibertate/games/net/MatchControllerNetworkTest.kt @@ -228,17 +228,78 @@ class MatchControllerNetworkTest { guestLeft.start() guestLeft.leave(Side.GUEST) guestLeft.settle() - assertEquals("Guest left the game.", guestLeft.host.abandoned.value) + assertEquals("The connection to Guest was lost.", guestLeft.host.abandoned.value) val hostLeft = chess() hostLeft.start() hostLeft.leave(Side.HOST) hostLeft.settle() - assertEquals("The host left the game.", hostLeft.client.abandoned.value) + assertEquals("The connection to the host was lost.", hostLeft.client.abandoned.value) // Nothing more can be played on a table with no host. assertEquals(emptyList(), hostLeft.client.legalMoves.value) } + // ----------------------------------------------------------------------- + // Coming back + // ----------------------------------------------------------------------- + + /** + * The bug this is here for: a phone locked when its turn came round could + * not play once it was unlocked again. + * + * A locked phone puts its Wi-Fi radio to sleep, and a link carrying nothing + * between one turn and the next is exactly what that drops. The game was + * never wrong — the host still had the position and the seat was still the + * player's — but the only route to it was gone, and nothing rebuilt it. The + * board came back on unlocking with the state it had before the turn + * arrived, and every tap on it went nowhere. + */ + @Test + fun `a guest whose link died can play again once it is back`() = runBlocking { + val table = chess() + table.start() + table.white("e2", "e4") + // Twenty replies to 1. e4 — the guest is on the clock. + assertEquals(20, table.client.legalMoves.value.size) + + table.dropGuestLink() + assertNotNull("A dead link has to stop the table", table.client.abandoned.value) + assertNotNull(table.host.abandoned.value) + + table.reconnectGuest() + + assertNull("The table is playable again and must not say otherwise", table.client.abandoned.value) + assertNull(table.host.abandoned.value) + assertNotNull("The guest came back to an empty board", table.client.state.value) + assertEquals( + "The guest is back but has nothing it is allowed to play", + 20, + table.client.legalMoves.value.size, + ) + + // And a move made on the new link is a move that actually happens. + table.black("e7", "e5") + assertNotNull(table.host.state.value?.board?.get(squareFromName("e5"))) + assertNull(table.host.state.value?.board?.get(squareFromName("e7"))) + } + + /** + * A guest coming back must not be treated as a guest arriving. The host end + * of the old link is the one that died; anything it has left to say about + * itself belongs to it and not to the seat, which is occupied again. + */ + @Test + fun `a guest coming back is not reported as having left`() = runBlocking { + val table = chess() + table.start() + table.dropGuestLink() + table.reconnectGuest() + table.settle() + + assertNull(table.host.abandoned.value) + assertNull(table.client.abandoned.value) + } + @Test fun `closing a link on purpose is not reported as the other player leaving`() = runBlocking { val table = chess() @@ -273,6 +334,11 @@ class MatchControllerNetworkTest { private val hostScope = CoroutineScope(Job() + Dispatchers.Default) private val clientScope = CoroutineScope(Job() + Dispatchers.Default) + /** Every socket a reconnection has opened since, so dispose can close them. */ + private val laterSockets = mutableListOf() + private val guestPeerId: String + private val guestSeat: Int + val hostLink: StreamConnection private val clientLink: StreamConnection val host: MatchController @@ -281,7 +347,8 @@ class MatchControllerNetworkTest { init { clientSocket.connect(InetSocketAddress("127.0.0.1", server.localPort), CONNECT_MILLIS) hostSocket = server.accept() - val guestPeerId = hostSocket.inetAddress?.hostAddress ?: "guest" + guestPeerId = hostSocket.inetAddress?.hostAddress ?: "guest" + guestSeat = (0 until seatCount).first { it != hostSeat } hostLink = StreamConnection( peerId = guestPeerId, @@ -332,11 +399,7 @@ class MatchControllerNetworkTest { ) host = controller(MatchController.Role.HOST, hostSeat, hostScope) - client = controller( - MatchController.Role.CLIENT, - (0 until seatCount).first { it != hostSeat }, - clientScope, - ) + client = controller(MatchController.Role.CLIENT, guestSeat, clientScope) // The lobby is already reading the guest's link when the game starts, // and goes on reading it for the whole match. Without this the test @@ -372,6 +435,54 @@ class MatchControllerNetworkTest { settle() } + /** + * The guest's link dies under it, as a sleeping radio drops one. + * + * The sockets are killed rather than the connections closed: a link + * either end let go of on purpose is a player leaving, which is a + * different thing entirely and is already tested above. + */ + suspend fun dropGuestLink() { + runCatching { hostSocket.close() } + runCatching { clientSocket.close() } + settle() + } + + /** + * The guest dials the host again, which is still listening on the same + * port, and both ends are handed the new link — the host from its lobby + * on the guest's Hello, the guest from its own once that has gone + * through. + */ + suspend fun reconnectGuest() { + val socket = Socket() + socket.connect(InetSocketAddress("127.0.0.1", server.localPort), CONNECT_MILLIS) + val accepted = server.accept() + laterSockets += socket + laterSockets += accepted + + val newHostLink = StreamConnection( + peerId = guestPeerId, + kind = TransportKind.LAN, + input = accepted.getInputStream(), + output = accepted.getOutputStream(), + scope = hostScope, + onClosed = { runCatching { accepted.close() } }, + ) + val newClientLink = StreamConnection( + peerId = "lan:host", + kind = TransportKind.LAN, + input = socket.getInputStream(), + output = socket.getOutputStream(), + scope = clientScope, + onClosed = { runCatching { socket.close() } }, + ) + + host.rebindPeer(guestSeat, newHostLink) + client.rebindHostLink(newClientLink) + settle() + } + /** Exactly what AppRoot.pop() and PlayScreen's onDispose do, in that order. */ fun leave(side: Side) { if (side == Side.HOST) { @@ -392,6 +503,7 @@ class MatchControllerNetworkTest { fun dispose() { runCatching { hostSocket.close() } runCatching { clientSocket.close() } + laterSockets.forEach { runCatching { it.close() } } runCatching { server.close() } runCatching { hostLink.close() } runCatching { clientLink.close() } diff --git a/app/src/test/java/org/prolibertate/games/net/StreamConnectionHeartbeatTest.kt b/app/src/test/java/org/prolibertate/games/net/StreamConnectionHeartbeatTest.kt new file mode 100644 index 0000000..6d64097 --- /dev/null +++ b/app/src/test/java/org/prolibertate/games/net/StreamConnectionHeartbeatTest.kt @@ -0,0 +1,181 @@ +package org.prolibertate.games.net + +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.Job +import kotlinx.coroutines.cancel +import kotlinx.coroutines.delay +import kotlinx.coroutines.launch +import kotlinx.coroutines.runBlocking +import org.junit.After +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertTrue +import org.junit.Test +import java.io.InputStream +import java.io.OutputStream +import java.net.InetSocketAddress +import java.net.ServerSocket +import java.net.Socket +import java.util.Collections +import java.util.concurrent.CountDownLatch + +/** + * The heartbeat, which exists because a link with nothing on it cannot be told + * from a link that is no longer there. + * + * Between one turn of a card game and the next, nothing at all crosses the + * socket. A phone whose screen has gone off puts its Wi-Fi radio into power + * save and, given long enough, the link goes with it — and TCP does not say so: + * the read simply never returns, and the next write disappears into a socket + * that goes nowhere. So the transport talks to itself, and treats a peer that + * has stopped answering as gone. + */ +class StreamConnectionHeartbeatTest { + + private val scopes = mutableListOf() + private val closeables = mutableListOf<() -> Unit>() + + @After + fun tearDown() { + closeables.forEach { runCatching { it() } } + scopes.forEach { it.cancel() } + } + + private fun scope(): CoroutineScope = + CoroutineScope(Job() + Dispatchers.Default).also { scopes += it } + + // ----------------------------------------------------------------------- + + /** + * The case the whole thing has to get right: two players thinking, nobody + * moving, for far longer than a link is allowed to be silent. Both ends + * answer each other's pings, so neither concludes the other has gone. + */ + @Test + fun `a quiet link between two live peers is left alone`() = runBlocking { + val server = ServerSocket(0) + val clientSocket = Socket() + closeables += { server.close() } + closeables += { clientSocket.close() } + + clientSocket.connect(InetSocketAddress("127.0.0.1", server.localPort), CONNECT_MILLIS) + val hostSocket = server.accept() + closeables += { hostSocket.close() } + + val hostScope = scope() + val clientScope = scope() + val a = connection("host", hostSocket.getInputStream(), hostSocket.getOutputStream(), hostScope) + val b = connection("guest", clientSocket.getInputStream(), clientSocket.getOutputStream(), clientScope) + + val heardByA = subscribe(a, hostScope) + val heardByB = subscribe(b, clientScope) + + // Several times the silence allowance, with not one game message sent. + delay(SILENCE_MILLIS * 4) + + assertTrue("The host's end gave up on a peer that was answering", a.isOpen) + assertTrue("The guest's end gave up on a peer that was answering", b.isOpen) + // And the pings are the transport's own business: nothing above it sees + // them, so a screen is not asked to redraw twice a second for nothing. + assertEquals(emptyList(), heardByA) + assertEquals(emptyList(), heardByB) + } + + /** + * A link that has died without saying so. The reader would wait on it for + * as long as the game lasted, which is what left a player tapping a board + * that answered nothing. + */ + @Test + fun `a peer that stops answering is given up on and reported`() = runBlocking { + val scope = scope() + val input = BlockingInputStream() + closeables += { input.close() } + + val link = connection("gone", input, DiscardingOutputStream(), scope) + val heard = subscribe(link, scope) + + // Long enough for the silence to be noticed, and then some. + delay(SILENCE_MILLIS * 3) + + assertFalse("A link nobody is answering on was left open", link.isOpen) + assertEquals( + "The screen was never told the link had gone", + listOf(Bye), + heard, + ) + } + + /** A link this end let go of is not somebody else going away. */ + @Test + fun `closing a link on purpose says nothing`() = runBlocking { + val scope = scope() + val input = BlockingInputStream() + closeables += { input.close() } + + val link = connection("ours", input, DiscardingOutputStream(), scope) + val heard = subscribe(link, scope) + + link.close() + delay(SILENCE_MILLIS) + + assertFalse(link.isOpen) + assertEquals(emptyList(), heard) + } + + // ----------------------------------------------------------------------- + + private fun connection( + peerId: String, + input: InputStream, + output: OutputStream, + scope: CoroutineScope, + ) = StreamConnection( + peerId = peerId, + kind = TransportKind.LAN, + input = input, + output = output, + scope = scope, + heartbeatMillis = HEARTBEAT_MILLIS, + silenceMillis = SILENCE_MILLIS, + ) + + /** Everything a screen would have been shown, in order. */ + private fun subscribe(link: Connection, scope: CoroutineScope): List { + val heard = Collections.synchronizedList(mutableListOf()) + scope.launch { link.incoming.collect { heard += it } } + return heard + } + + /** + * A stream that has nothing to say and never will, which is what a link + * whose far end has vanished looks like from here. Blocks until it is + * closed, then fails the read the way a closed socket does. + */ + private class BlockingInputStream : InputStream() { + private val gate = CountDownLatch(1) + + override fun read(): Int { + gate.await() + throw java.io.IOException("stream closed") + } + + override fun close() = gate.countDown() + } + + private class DiscardingOutputStream : OutputStream() { + override fun write(b: Int) = Unit + } + + private companion object { + const val CONNECT_MILLIS = 2_000 + + /** + * The real intervals are seconds; these are the same arrangement run + * fast, so the tests take a moment rather than a minute. + */ + const val HEARTBEAT_MILLIS = 40L + const val SILENCE_MILLIS = 200L + } +}