diff --git a/app/src/main/java/com/itsaky/androidide/lsp/BreakpointHandler.kt b/app/src/main/java/com/itsaky/androidide/lsp/BreakpointHandler.kt index a82ec8b618..922cbadb44 100644 --- a/app/src/main/java/com/itsaky/androidide/lsp/BreakpointHandler.kt +++ b/app/src/main/java/com/itsaky/androidide/lsp/BreakpointHandler.kt @@ -15,11 +15,12 @@ import com.itsaky.androidide.models.Position import com.itsaky.androidide.projects.IProjectManager import com.itsaky.androidide.repositories.BreakpointRepository import com.itsaky.androidide.repositories.StoredBreakpointsType +import kotlinx.coroutines.CoroutineExceptionHandler import kotlinx.coroutines.CoroutineScope -import kotlinx.coroutines.DelicateCoroutinesApi import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.Job +import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableStateFlow @@ -27,7 +28,6 @@ import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow import kotlinx.coroutines.flow.update import kotlinx.coroutines.launch -import kotlinx.coroutines.newSingleThreadContext import org.slf4j.LoggerFactory import java.io.File import java.util.TreeMap @@ -36,12 +36,12 @@ import java.util.concurrent.atomic.AtomicReference private interface BreakpointEvent { data class DocChange( - val event: DocumentChangeEvent + val event: DocumentChangeEvent, ) : BreakpointEvent data class Toggle( val file: File, - val line: Int + val line: Int, ) : BreakpointEvent object Save : BreakpointEvent @@ -49,25 +49,46 @@ private interface BreakpointEvent { private data class BpState( val positional: Table, - val method: Table + val method: Table, ) { companion object { val EMPTY = BpState(ImmutableTable.of(), ImmutableTable.of()) } } -class BreakpointHandler { +class BreakpointHandler : AutoCloseable { + // limitedParallelism(1) gives the sequential confinement this class relies on without owning a + // thread, so there is nothing to close and nothing to leak per editor session (ADFA-5375). + // Without a handler an failure here escapes to the process-wide uncaught handler and takes the + // app down; the breakpoint consumer is not worth a crash. + private val exceptionHandler = + CoroutineExceptionHandler { _, error -> + logger.error("Unhandled error while processing breakpoint events", error) + } + + @OptIn(ExperimentalCoroutinesApi::class) + private val scope = + CoroutineScope(Dispatchers.IO.limitedParallelism(1) + SupervisorJob() + exceptionHandler) + + // Persistence deliberately outlives [close]: a debounced save must still land when the editor + // goes away, which is exactly when the user is most likely to lose a just-added breakpoint. + private val saveScope = CoroutineScope(Dispatchers.IO + SupervisorJob()) - @OptIn(ExperimentalCoroutinesApi::class, DelicateCoroutinesApi::class) - private val scope = CoroutineScope(newSingleThreadContext("BreakpointHandler")) private val events = Channel(capacity = Channel.UNLIMITED) private val _highlightedLocation = MutableStateFlow?>(null) + + @Volatile private var onSetBreakpoints: ((List) -> Unit)? = null private val listeners = CopyOnWriteArrayList() private val stateRef = AtomicReference(BpState.EMPTY) + + @Volatile private var saveJob: Job? = null + @Volatile + private var closed = false + val highlightedLocationState: StateFlow?> get() = _highlightedLocation.asStateFlow() @@ -75,16 +96,20 @@ class BreakpointHandler { get() = highlightedLocationState.value val allBreakpoints: List - get() = stateRef.get().let { s -> - buildList { - addAll(s.positional.values()) - addAll(s.method.values()) + get() = + stateRef.get().let { s -> + buildList { + addAll(s.positional.values()) + addAll(s.method.values()) + } } - } companion object { private val logger = LoggerFactory.getLogger(BreakpointHandler::class.java) + /** How long a burst of breakpoint edits is coalesced before it reaches disk. */ + private const val SAVE_DEBOUNCE_MS = 1000L + @VisibleForTesting internal fun computeNewBreakpointPosition( line: Int, @@ -97,15 +122,12 @@ class BreakpointHandler { var newColumn = column if (changeType == ChangeType.INSERT) { - // insertion before breakpoint line if (line > start.line) { // shift down newLine += end.line - start.line - } - - // insertion on breakpoint line, after start column - else if (line == start.line && column > start.column) { + } else if (line == start.line && column > start.column) { + // insertion on breakpoint line, after start column if (start.line == end.line) { // same line insertion, shift column right newColumn += end.column - start.column @@ -121,25 +143,20 @@ class BreakpointHandler { if (line > end.line) { // shift up newLine -= end.line - start.line - } - - // breakpoint after last line of deletion, after end column - else if (line == end.line && column > end.column) { + } else if (line == end.line && column > end.column) { + // breakpoint after last line of deletion, after end column if (start.line == end.line) { // Single-line deletion - newColumn -= (end.column - start.column); + newColumn -= (end.column - start.column) } else { // Multi-line deletion newLine = start.line - newColumn = start.column + (column - end.column); + newColumn = start.column + (column - end.column) } - } - - // breakpoint within deleted range - else if ((line > start.line || (line == start.line && column >= start.column)) - && (line < end.line || (line == end.line && column <= end.column)) + } else if ((line > start.line || (line == start.line && column >= start.column)) && + (line < end.line || (line == end.line && column <= end.column)) ) { - // mark for deletion + // breakpoint within deleted range: mark for deletion newLine = -1 newColumn = -1 } @@ -149,7 +166,10 @@ class BreakpointHandler { } } - fun highlightLocation(file: String, line: Int) { + fun highlightLocation( + file: String, + line: Int, + ) { this._highlightedLocation.update { file to line } notifyHighlighted(file, line) } @@ -175,6 +195,39 @@ class BreakpointHandler { } } + /** + * Detaches the handler from its owner. Closing [events] lets the consumer drain what is already + * queued and then finish on its own, rather than cancelling it and dropping those edits; the + * scope owns no thread, so an idle scope costs nothing. Listeners are dropped first so the + * drained events cannot call back into a destroyed editor. + */ + override fun close() { + if (closed) { + return + } + closed = true + + unhighlightHighlightedLocation() + listeners.clear() + onSetBreakpoints = null + events.close() + flushPendingSave() + } + + /** + * Writes a debounced save immediately instead of waiting out the remaining delay. Without this, + * a breakpoint toggled in the last [SAVE_DEBOUNCE_MS] before the editor closes is lost. + */ + private fun flushPendingSave() { + val pending = saveJob ?: return + if (!pending.isActive) { + return + } + + pending.cancel() + saveJob = saveScope.launch { writeBreakpoints() } + } + fun addListener(listener: EventListener) { if (listeners.contains(listener)) { logger.warn("listener {} is already added", listener) @@ -189,7 +242,8 @@ class BreakpointHandler { } suspend fun positionalBreakpointsInFile(file: File): List = - BreakpointRepository.getStoredBreakpoints(IProjectManager.getInstance().projectDir) + BreakpointRepository + .getStoredBreakpoints(IProjectManager.getInstance().projectDir) .mapNotNull { breakpoint -> if (breakpoint.source.path != file.absolutePath || breakpoint !is PositionalBreakpoint) { return@mapNotNull null @@ -198,12 +252,30 @@ class BreakpointHandler { breakpoint } - suspend fun change(event: DocumentChangeEvent) { - events.send(BreakpointEvent.DocChange(event)) + fun change(event: DocumentChangeEvent) { + offer(BreakpointEvent.DocChange(event), "document change") } - suspend fun toggle(file: File, line: Int) { - events.send(BreakpointEvent.Toggle(file, line)) + fun toggle( + file: File, + line: Int, + ) { + offer(BreakpointEvent.Toggle(file, line), "toggle at $file:$line") + } + + /** + * The channel is unbounded, so the only way this fails is a closed handler - which is a race a + * caller cannot avoid, not an error worth throwing into its scope. `send` here would raise + * ClosedSendChannelException on a scope with no handler, and the IDE's uncaught handler turns + * that into a process exit. + */ + private fun offer( + event: BreakpointEvent, + description: String, + ) { + if (events.trySend(event).isFailure) { + logger.debug("Dropping {}: the breakpoint handler is closed", description) + } } private fun refreshBreakpoints(loadedBreakpoints: StoredBreakpointsType) { @@ -218,23 +290,25 @@ class BreakpointHandler { } } - val snap = BpState( - positional = ImmutableTable.copyOf(newPos), - method = ImmutableTable.copyOf(newMethod) - ) + val snap = + BpState( + positional = ImmutableTable.copyOf(newPos), + method = ImmutableTable.copyOf(newMethod), + ) stateRef.set(snap) } - private fun process(event: BreakpointEvent) = runCatching { - when (event) { - is BreakpointEvent.DocChange -> onChange(event) - is BreakpointEvent.Toggle -> onToggle(event) - is BreakpointEvent.Save -> onSave() + private fun process(event: BreakpointEvent) = + runCatching { + when (event) { + is BreakpointEvent.DocChange -> onChange(event) + is BreakpointEvent.Toggle -> onToggle(event) + is BreakpointEvent.Save -> onSave() + } + }.onFailure { err -> + logger.error("Failed to handle event {}", event.javaClass.simpleName, err) } - }.onFailure { err -> - logger.error("Failed to handle event {}", event.javaClass.simpleName, err) - } private fun onChange(event: BreakpointEvent.DocChange) { val ev = event.event @@ -243,11 +317,16 @@ class BreakpointHandler { val path = ev.file.toRealPath().toString() logger.debug( - "change({}): range={},{}-{},{}", when (ev.changeType) { + "change({}): range={},{}-{},{}", + when (ev.changeType) { ChangeType.NEW_TEXT -> "new text" ChangeType.DELETE -> "delete" ChangeType.INSERT -> "insert" - }, start.line, start.column, end.line, end.column + }, + start.line, + start.column, + end.line, + end.column, ) if (end.line - start.line == 0) { @@ -284,13 +363,14 @@ class BreakpointHandler { for ((_, bp) in fileBreakpoints) { val line = bp.line val column = bp.column - val (newLine, newColumn) = computeNewBreakpointPosition( - line, - column, - start, - end, - ev.changeType - ) + val (newLine, newColumn) = + computeNewBreakpointPosition( + line, + column, + start, + end, + ev.changeType, + ) if (newLine == line && newColumn == column) { logger.debug("keep breakpoint at line {} in file {}", line, path) @@ -324,10 +404,11 @@ class BreakpointHandler { val (file, line) = event val path = file.canonicalPath - val breakpoint = PositionalBreakpoint( - source = Source(path = file.absolutePath, name = file.name), - line = line - ) + val breakpoint = + PositionalBreakpoint( + source = Source(path = file.absolutePath, name = file.name), + line = line, + ) val current = stateRef.get() val newPos = HashBasedTable.create(current.positional) @@ -353,51 +434,77 @@ class BreakpointHandler { } private fun onSave() { + if (closed) { + return + } + saveJob?.cancel() - saveJob = scope.launch(Dispatchers.IO) { - delay(1000) - val snap = stateRef.get() - val breakpointsToSave = buildList { + saveJob = + saveScope.launch { + delay(SAVE_DEBOUNCE_MS) + writeBreakpoints() + } + } + + private suspend fun writeBreakpoints() { + val snap = stateRef.get() + val breakpointsToSave = + buildList { addAll(snap.positional.values()) addAll(snap.method.values()) } - BreakpointRepository.saveBreakpoints( - projectDir = IProjectManager.getInstance().projectDir, - breakpoints = breakpointsToSave - ) - logger.debug("Breakpoints saved to disk.") - } + BreakpointRepository.saveBreakpoints( + projectDir = IProjectManager.getInstance().projectDir, + breakpoints = breakpointsToSave, + ) + logger.debug("Breakpoints saved to disk.") } private fun notifyBreakpointsUpdated(newBreakpoints: List) { onSetBreakpoints?.invoke(newBreakpoints) } - private fun notifyAdded(file: String, line: Int) { + private fun notifyAdded( + file: String, + line: Int, + ) { for (listener in listeners) { listener.onAddBreakpoint(file, line) } } - private fun notifyRemoved(file: String, line: Int) { + private fun notifyRemoved( + file: String, + line: Int, + ) { for (listener in listeners) { listener.onRemoveBreakpoint(file, line) } } - private fun notifyToggled(file: String, line: Int) { + private fun notifyToggled( + file: String, + line: Int, + ) { for (listener in listeners) { listener.onToggle(file, line) } } - private fun notifyMoved(file: String, oldLine: Int, newLine: Int) { + private fun notifyMoved( + file: String, + oldLine: Int, + newLine: Int, + ) { for (listener in listeners) { listener.onMoveBreakpoint(file, oldLine, newLine) } } - private fun notifyHighlighted(file: String, line: Int) { + private fun notifyHighlighted( + file: String, + line: Int, + ) { for (listener in listeners) { listener.onHighlightLine(file, line) } @@ -410,11 +517,32 @@ class BreakpointHandler { } interface EventListener { - fun onAddBreakpoint(file: String, line: Int) {} - fun onRemoveBreakpoint(file: String, line: Int) {} - fun onToggle(file: String, line: Int) {} - fun onMoveBreakpoint(file: String, oldLine: Int, newLine: Int) {} - fun onHighlightLine(file: String, line: Int) {} + fun onAddBreakpoint( + file: String, + line: Int, + ) {} + + fun onRemoveBreakpoint( + file: String, + line: Int, + ) {} + + fun onToggle( + file: String, + line: Int, + ) {} + + fun onMoveBreakpoint( + file: String, + oldLine: Int, + newLine: Int, + ) {} + + fun onHighlightLine( + file: String, + line: Int, + ) {} + fun onUnhighlight() } -} \ No newline at end of file +} diff --git a/app/src/main/java/com/itsaky/androidide/lsp/IDEDebugClientImpl.kt b/app/src/main/java/com/itsaky/androidide/lsp/IDEDebugClientImpl.kt index 83f35c74a0..4f6b8098ed 100644 --- a/app/src/main/java/com/itsaky/androidide/lsp/IDEDebugClientImpl.kt +++ b/app/src/main/java/com/itsaky/androidide/lsp/IDEDebugClientImpl.kt @@ -2,6 +2,7 @@ package com.itsaky.androidide.lsp import android.annotation.SuppressLint import android.widget.RemoteViews.RemoteView +import androidx.annotation.VisibleForTesting import com.itsaky.androidide.eventbus.events.EventReceiver import com.itsaky.androidide.eventbus.events.editor.DocumentChangeEvent import com.itsaky.androidide.lookup.Lookup @@ -19,15 +20,17 @@ import com.itsaky.androidide.lsp.debug.model.StepType import com.itsaky.androidide.lsp.debug.model.ThreadListRequestParams import com.itsaky.androidide.models.Position import com.itsaky.androidide.models.Range +import com.itsaky.androidide.tasks.cancelIfActive import com.itsaky.androidide.viewmodel.DebuggerConnectionState import com.itsaky.androidide.viewmodel.DebuggerViewModel +import kotlinx.coroutines.CoroutineExceptionHandler import kotlinx.coroutines.CoroutineScope -import kotlinx.coroutines.DelicateCoroutinesApi import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.isActive import kotlinx.coroutines.launch -import kotlinx.coroutines.newFixedThreadPoolContext import kotlinx.coroutines.withContext import org.greenrobot.eventbus.Subscribe import org.greenrobot.eventbus.ThreadMode @@ -39,303 +42,397 @@ import java.util.concurrent.CopyOnWriteArraySet * @author Akash Yadav */ class IDEDebugClientImpl( - private val viewModel: DebuggerViewModel, -) : IDebugClient, IDebugEventHandler, EventReceiver { - - private val logger = LoggerFactory.getLogger(IDEDebugClientImpl::class.java) - - @OptIn(DelicateCoroutinesApi::class) - private val clientContext = newFixedThreadPoolContext(4, "IDEDebugClient") - private val clientScope = CoroutineScope(clientContext + SupervisorJob()) - private val clients = CopyOnWriteArraySet() - val breakpoints = BreakpointHandler() - - companion object { - @JvmStatic - fun getInstance() = Lookup.getDefault().lookup(IDEDebugClientImpl::class.java) - - @JvmStatic - fun requireInstance() = checkNotNull(getInstance()) { - "Cannot lookup IDEDebugClientImpl" - } - } - - val connectionStateFlow: StateFlow - get() = viewModel.connectionState - - var connectionState: DebuggerConnectionState - get() = viewModel.connectionState.value - private set(value) { - logger.debug("move to connection state: {}", value) - viewModel.setConnectionState(value) - } - - val debugeePackage: String - get() = viewModel.debugeePackage - - internal val requireClient: RemoteClient - get() = checkNotNull(clientOrNull) - - internal val clientOrNull: RemoteClient? - get() = clients.firstOrNull() - - /** - * Returns true if the client is connected. - * - * @return `true` if the client is connected, `false` otherwise. - */ - fun isVmConnected() = connectionState >= DebuggerConnectionState.ATTACHED - - /** - * Returns true if the client is connected and suspended. - * - * The VM may or may not be able to view/alter its state. Check [connectionState] - * to get the actual state. - * - * @return `true` if the client is connected and suspended, `false` otherwise. - */ - fun isVmSuspended() = connectionState >= DebuggerConnectionState.SUSPENDED - - fun suspendVm() = withClient("suspend vm") { client -> - if (!client.capabilities.suspensionSupport) { - logger.error("Remote client does not support suspending") - return@withClient - } - - if (isVmSuspended()) { - logger.warn("Ignoring attempt to suspend VM when it is already suspended") - return@withClient - } - - logger.debug("suspending client: {}", client.name) - clientScope.launch { - if (client.adapter.suspendClient(client)) { - connectionState = DebuggerConnectionState.SUSPENDED - updateThreadInfo(client) - } - } - } - - fun resumeVm() = withClient("resume vm") { client -> - if (!client.capabilities.suspensionSupport) { - logger.error("Remote client does not support resuming") - return@withClient - } - - if (!isVmSuspended()) { - logger.warn("Ignoring attempt to resume VM when it is not suspended") - return@withClient - } - - logger.debug("resuming client: {}", client.name) - clientScope.launch { - if (client.adapter.resumeClient(client)) { - connectionState = DebuggerConnectionState.ATTACHED - updateThreadInfo(client) - } - } - } - - fun killVm() = withClient("kill vm") { client -> - if (!client.capabilities.killSupport) { - logger.error("Remote client does not support killing debug application") - return@withClient - } - - logger.debug("killing client: {}", client.name) - clientScope.launch { client.adapter.killClient(client) } - } - - fun stepOver() = doStep(type = StepType.Over) - fun stepInto() = doStep(type = StepType.Into) - fun stepOut() = doStep(type = StepType.Out) - - private inline fun withClient(action: String, block: (RemoteClient) -> Unit) { - clientOrNull?.also(block) - ?: logger.error("Cannot perform $action action. Not connected to a remote client.") - } - - private fun doStep( - type: StepType, - countFilter: Int = 1 - ) = withClient("step $type") { client -> - if (!client.capabilities.stepSupport) { - logger.error("Remote client does not support stepping") - return@withClient - } - - clientScope.launch { - val params = StepRequestParams( - remoteClient = client, - type = type, - countFilter = countFilter - ) - - val response = client.adapter.step(params) - if (response.result != StepResult.Success) { - logger.error("Failed to perform step action, result={}", response.result) - } - } - } - - init { - register() - breakpoints.begin { breakpoints -> - - // if we're already connected to a client, update the client as well - clientOrNull?.also { client -> - if (!client.capabilities.breakpointSupport) { - logger.error("Remote client does not support breakpoints") - return@also - } - - clientScope.launch { - clientOrNull?.also { client -> - val response = client.adapter.setBreakpoints( - BreakpointRequest( - remoteClient = client, - breakpoints = breakpoints - ) - ) - - logger.debug("breakpoint result: {}", response.results) - } ?: logger.info("client ${client.name} disconnected while updating breakpoints") - } - } ?: logger.info("deferring breakpoint update, no clients connected") - } - } - - @SuppressLint("ImplicitSamInstance") - @Suppress("UNUSED") - @Subscribe(threadMode = ThreadMode.ASYNC) - fun onContentChange(event: DocumentChangeEvent) { - clientScope.launch { breakpoints.change(event) } - } - - fun toggleBreakpoint(file: File, line: Int) { - clientScope.launch { breakpoints.toggle(file, line) } - } - - override fun onBreakpointHit(event: BreakpointHitEvent) { - logger.debug("onBreakpointHit: {}", event) - - clientScope.launch { - connectionState = DebuggerConnectionState.AWAITING_BREAKPOINT - updateThreadInfo(event.remoteClient, event.threadId) - - openLocation(event) - } - } - - override fun onStep(event: StepEvent) { - logger.debug("onStep: {}", event) - - clientScope.launch { - updateThreadInfo(event.remoteClient, event.threadId) - connectionState = DebuggerConnectionState.AWAITING_BREAKPOINT - - openLocation(event) - } - } - - override fun onAttach(client: RemoteClient) { - logger.debug("onAttach: client={}", client) - - check(client !in clients) { - "Already attached to client" - } - - clients += client - connectionState = DebuggerConnectionState.ATTACHED - breakpoints.unhighlightHighlightedLocation() - - clientScope.launch { - updateThreadInfo(client) - - val breakpoints = breakpoints.allBreakpoints - client.adapter.setBreakpoints( - BreakpointRequest( - remoteClient = client, - breakpoints = breakpoints - ) - ) - } - } - - override fun onDisconnect(client: RemoteClient) { - logger.debug("onDisconnect: client={}", client) - breakpoints.unhighlightHighlightedLocation() - clients -= client - connectionState = DebuggerConnectionState.DETACHED - clientScope.launch { updateThreadInfo(client) } - } - - suspend fun updateThreadInfo( - client: RemoteClient, - selectedThreadId: String? = null, - ) { - val adapter = client.adapter - var selectedThreadIndex = -1 - val threads = when (connectionState) { - DebuggerConnectionState.DETACHED, - DebuggerConnectionState.ATTACHED -> emptyList() - - DebuggerConnectionState.SUSPENDED, - DebuggerConnectionState.AWAITING_BREAKPOINT -> { - val threadResponse = adapter.allThreads( - ThreadListRequestParams( - remoteClient = client - ) - ) - - val threads = threadResponse.threads - if (threads.isEmpty()) { - logger.error("Failed to get info about active threads in VM: {}", client.name) - } else if (selectedThreadId != null) { - selectedThreadIndex = threads.indexOfFirst { - it.descriptor().id == selectedThreadId - } - } - - threads - } - } - - if (threads.isNotEmpty() && selectedThreadIndex < 0) { - selectedThreadIndex = 0 - } - - viewModel.setThreads(threads, selectedThreadIndex) - } - - private suspend fun openLocation(event: LocatableEvent) = openLocation(event.location) - - private suspend fun openLocation(location: Location) { - val file = location.source.path - val position = Position(location.line, 0) - - if (!IDELanguageClientImpl.isInitialized()) { - logger.error("Cannot open {}:{} because language client is not initialized", file, position.line) - return - } - - val activity = IDELanguageClientImpl.getInstance().activity - - if (activity == null) { - logger.error("Cannot open {}:{} because activity is null", file, position.line) - return - } - - breakpoints.highlightLocation(file, position.line) - - withContext(Dispatchers.Main.immediate) { - activity.openFileAndSelect( - file = File(file), - selection = Range( - start = position, - end = position - ) - ) - } - } -} \ No newline at end of file + private val viewModel: DebuggerViewModel, +) : IDebugClient, + IDebugEventHandler, + EventReceiver, + AutoCloseable { + private val logger = LoggerFactory.getLogger(IDEDebugClientImpl::class.java) + + // limitedParallelism bounds concurrency the way a fixed pool did, but owns no thread, so there + // is nothing to close and nothing to leak per editor session (ADFA-5375). + @OptIn(ExperimentalCoroutinesApi::class) + private val clientContext = Dispatchers.IO.limitedParallelism(4) + private val clientScope = + CoroutineScope( + clientContext + SupervisorJob() + + CoroutineExceptionHandler { _, error -> + // The IDE's uncaught handler exits the process; a failed debugger call must not. + logger.error("Unhandled error in debug client", error) + }, + ) + private val clients = CopyOnWriteArraySet() + val breakpoints = BreakpointHandler() + + @Volatile + private var closed = false + + companion object { + @JvmStatic + fun getInstance() = Lookup.getDefault().lookup(IDEDebugClientImpl::class.java) + + @JvmStatic + fun requireInstance() = + checkNotNull(getInstance()) { + "Cannot lookup IDEDebugClientImpl" + } + } + + val connectionStateFlow: StateFlow + get() = viewModel.connectionState + + var connectionState: DebuggerConnectionState + get() = viewModel.connectionState.value + private set(value) { + logger.debug("move to connection state: {}", value) + viewModel.setConnectionState(value) + } + + val debugeePackage: String + get() = viewModel.debugeePackage + + internal val requireClient: RemoteClient + get() = checkNotNull(clientOrNull) + + internal val clientOrNull: RemoteClient? + get() = clients.firstOrNull() + + /** + * Returns true if the client is connected. + * + * @return `true` if the client is connected, `false` otherwise. + */ + fun isVmConnected() = connectionState >= DebuggerConnectionState.ATTACHED + + /** + * Returns true if the client is connected and suspended. + * + * The VM may or may not be able to view/alter its state. Check [connectionState] + * to get the actual state. + * + * @return `true` if the client is connected and suspended, `false` otherwise. + */ + fun isVmSuspended() = connectionState >= DebuggerConnectionState.SUSPENDED + + fun suspendVm() = + withClient("suspend vm") { client -> + if (!client.capabilities.suspensionSupport) { + logger.error("Remote client does not support suspending") + return@withClient + } + + if (isVmSuspended()) { + logger.warn("Ignoring attempt to suspend VM when it is already suspended") + return@withClient + } + + logger.debug("suspending client: {}", client.name) + clientScope.launch { + if (client.adapter.suspendClient(client)) { + connectionState = DebuggerConnectionState.SUSPENDED + updateThreadInfo(client) + } + } + } + + fun resumeVm() = + withClient("resume vm") { client -> + if (!client.capabilities.suspensionSupport) { + logger.error("Remote client does not support resuming") + return@withClient + } + + if (!isVmSuspended()) { + logger.warn("Ignoring attempt to resume VM when it is not suspended") + return@withClient + } + + logger.debug("resuming client: {}", client.name) + clientScope.launch { + if (client.adapter.resumeClient(client)) { + connectionState = DebuggerConnectionState.ATTACHED + updateThreadInfo(client) + } + } + } + + fun killVm() = + withClient("kill vm") { client -> + if (!client.capabilities.killSupport) { + logger.error("Remote client does not support killing debug application") + return@withClient + } + + logger.debug("killing client: {}", client.name) + clientScope.launch { client.adapter.killClient(client) } + } + + fun stepOver() = doStep(type = StepType.Over) + + fun stepInto() = doStep(type = StepType.Into) + + fun stepOut() = doStep(type = StepType.Out) + + private inline fun withClient( + action: String, + block: (RemoteClient) -> Unit, + ) { + if (isClosed(action)) { + return + } + + clientOrNull?.also(block) + ?: logger.error("Cannot perform $action action. Not connected to a remote client.") + } + + private fun doStep( + type: StepType, + countFilter: Int = 1, + ) = withClient("step $type") { client -> + if (!client.capabilities.stepSupport) { + logger.error("Remote client does not support stepping") + return@withClient + } + + clientScope.launch { + val params = + StepRequestParams( + remoteClient = client, + type = type, + countFilter = countFilter, + ) + + val response = client.adapter.step(params) + if (response.result != StepResult.Success) { + logger.error("Failed to perform step action, result={}", response.result) + } + } + } + + init { + register() + breakpoints.begin { breakpoints -> + + // if we're already connected to a client, update the client as well + clientOrNull?.also { client -> + if (!client.capabilities.breakpointSupport) { + logger.error("Remote client does not support breakpoints") + return@also + } + + clientScope.launch { + clientOrNull?.also { client -> + val response = + client.adapter.setBreakpoints( + BreakpointRequest( + remoteClient = client, + breakpoints = breakpoints, + ), + ) + + logger.debug("breakpoint result: {}", response.results) + } ?: logger.info("client ${client.name} disconnected while updating breakpoints") + } + } ?: logger.info("deferring breakpoint update, no clients connected") + } + } + + @SuppressLint("ImplicitSamInstance") + @Suppress("UNUSED") + @Subscribe(threadMode = ThreadMode.ASYNC) + fun onContentChange(event: DocumentChangeEvent) { + if (isClosed("document change")) { + return + } + + clientScope.launch { breakpoints.change(event) } + } + + fun toggleBreakpoint( + file: File, + line: Int, + ) { + if (isClosed("toggle breakpoint")) { + return + } + + clientScope.launch { breakpoints.toggle(file, line) } + } + + override fun onBreakpointHit(event: BreakpointHitEvent) { + logger.debug("onBreakpointHit: {}", event) + if (isClosed("breakpoint hit")) { + return + } + + clientScope.launch { + connectionState = DebuggerConnectionState.AWAITING_BREAKPOINT + updateThreadInfo(event.remoteClient, event.threadId) + + openLocation(event) + } + } + + override fun onStep(event: StepEvent) { + logger.debug("onStep: {}", event) + if (isClosed("step")) { + return + } + + clientScope.launch { + updateThreadInfo(event.remoteClient, event.threadId) + connectionState = DebuggerConnectionState.AWAITING_BREAKPOINT + + openLocation(event) + } + } + + override fun onAttach(client: RemoteClient) { + logger.debug("onAttach: client={}", client) + if (isClosed("attach")) { + return + } + + check(client !in clients) { + "Already attached to client" + } + + clients += client + connectionState = DebuggerConnectionState.ATTACHED + breakpoints.unhighlightHighlightedLocation() + + clientScope.launch { + updateThreadInfo(client) + + val breakpoints = breakpoints.allBreakpoints + client.adapter.setBreakpoints( + BreakpointRequest( + remoteClient = client, + breakpoints = breakpoints, + ), + ) + } + } + + override fun onDisconnect(client: RemoteClient) { + logger.debug("onDisconnect: client={}", client) + if (isClosed("disconnect")) { + return + } + + breakpoints.unhighlightHighlightedLocation() + clients -= client + connectionState = DebuggerConnectionState.DETACHED + clientScope.launch { updateThreadInfo(client) } + } + + suspend fun updateThreadInfo( + client: RemoteClient, + selectedThreadId: String? = null, + ) { + val adapter = client.adapter + var selectedThreadIndex = -1 + val threads = + when (connectionState) { + DebuggerConnectionState.DETACHED, + DebuggerConnectionState.ATTACHED, + -> { + emptyList() + } + + DebuggerConnectionState.SUSPENDED, + DebuggerConnectionState.AWAITING_BREAKPOINT, + -> { + val threadResponse = + adapter.allThreads( + ThreadListRequestParams( + remoteClient = client, + ), + ) + + val threads = threadResponse.threads + if (threads.isEmpty()) { + logger.error("Failed to get info about active threads in VM: {}", client.name) + } else if (selectedThreadId != null) { + selectedThreadIndex = + threads.indexOfFirst { + it.descriptor().id == selectedThreadId + } + } + + threads + } + } + + if (threads.isNotEmpty() && selectedThreadIndex < 0) { + selectedThreadIndex = 0 + } + + viewModel.setThreads(threads, selectedThreadIndex) + } + + /** + * Detaches from EventBus and cancels in-flight work, so nothing keeps calling into a + * [DebuggerViewModel] that has already been cleared. The client is unusable afterwards; every + * entry point below turns into a logged no-op. + */ + override fun close() { + if (closed) { + return + } + closed = true + + unregister() + clientScope.cancelIfActive("IDEDebugClientImpl closed") + breakpoints.close() + clients.clear() + } + + /** Whether in-flight work is still allowed; false once [close] has run. */ + @VisibleForTesting + internal val isClientScopeActive: Boolean + get() = clientScope.isActive + + /** + * Guards the entry points reachable from a stale reference or from the JDWP listener thread, + * which is not stopped by cancelling [clientScope]. + */ + private fun isClosed(action: String): Boolean { + if (closed) { + logger.warn("Ignoring {}: the debug client is closed", action) + } + return closed + } + + private suspend fun openLocation(event: LocatableEvent) = openLocation(event.location) + + private suspend fun openLocation(location: Location) { + val file = location.source.path + val position = Position(location.line, 0) + + if (!IDELanguageClientImpl.isInitialized()) { + logger.error("Cannot open {}:{} because language client is not initialized", file, position.line) + return + } + + val activity = IDELanguageClientImpl.getInstance().activity + + if (activity == null) { + logger.error("Cannot open {}:{} because activity is null", file, position.line) + return + } + + breakpoints.highlightLocation(file, position.line) + + withContext(Dispatchers.Main.immediate) { + activity.openFileAndSelect( + file = File(file), + selection = + Range( + start = position, + end = position, + ), + ) + } + } +} diff --git a/app/src/main/java/com/itsaky/androidide/utils/MemoryUsageWatcher.kt b/app/src/main/java/com/itsaky/androidide/utils/MemoryUsageWatcher.kt index 0e531ae964..5fe464066e 100644 --- a/app/src/main/java/com/itsaky/androidide/utils/MemoryUsageWatcher.kt +++ b/app/src/main/java/com/itsaky/androidide/utils/MemoryUsageWatcher.kt @@ -27,13 +27,11 @@ import com.itsaky.androidide.app.BaseApplication import com.itsaky.androidide.tasks.cancelIfActive import com.termux.shared.reflection.ReflectionUtils import kotlinx.coroutines.CoroutineScope -import kotlinx.coroutines.DelicateCoroutinesApi import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.delay import kotlinx.coroutines.launch -import kotlinx.coroutines.newSingleThreadContext import kotlinx.coroutines.withContext import org.slf4j.LoggerFactory import java.util.concurrent.ConcurrentHashMap @@ -48,8 +46,10 @@ import java.util.concurrent.atomic.AtomicBoolean class MemoryUsageWatcher( private val updateInterval: Long = DEFAULT_UPDATE_INTERVAL, ) { - @OptIn(ExperimentalCoroutinesApi::class, DelicateCoroutinesApi::class) - private val coroutineDispatcher = newSingleThreadContext("MemoryUsageWatcher") + // One watcher exists per editor activity, so a thread-owning dispatcher here leaked a thread per + // destroyed editor; limitedParallelism(1) keeps the sequencing and owns nothing (ADFA-5396). + @OptIn(ExperimentalCoroutinesApi::class) + private val coroutineDispatcher = Dispatchers.IO.limitedParallelism(1) private val coroutineScope = CoroutineScope(coroutineDispatcher) private val memoryUsage = ConcurrentHashMap() private val watching = AtomicBoolean(false) diff --git a/app/src/main/java/com/itsaky/androidide/viewmodel/DebuggerViewModel.kt b/app/src/main/java/com/itsaky/androidide/viewmodel/DebuggerViewModel.kt index 731cc9a9bf..5c83932aae 100644 --- a/app/src/main/java/com/itsaky/androidide/viewmodel/DebuggerViewModel.kt +++ b/app/src/main/java/com/itsaky/androidide/viewmodel/DebuggerViewModel.kt @@ -113,77 +113,91 @@ class DebuggerViewModel : ViewModel() { _debugeePackage.update { value } } - val allThreads = state.map { - logger.debug("Updating all threads") - it.threads - }.stateIn( - scope = viewModelScope, - started = SharingStarted.Eagerly, - initialValue = emptyList(), - ) - - val selectedThread = state - .map { state -> - state.selectedThread to state.threadIndex - }.stateIn( - scope = viewModelScope, - started = SharingStarted.Eagerly, - initialValue = null to -1, - ) + val allThreads = + state + .map { + logger.debug("Updating all threads") + it.threads + }.stateIn( + scope = viewModelScope, + started = SharingStarted.Eagerly, + initialValue = emptyList(), + ) - val allFrames = selectedThread - .map { (thread, _) -> - thread?.getFrames() ?: emptyList() - }.stateIn( - scope = viewModelScope, - started = SharingStarted.Eagerly, - initialValue = emptyList(), - ) + val selectedThread = + state + .map { state -> + state.selectedThread to state.threadIndex + }.stateIn( + scope = viewModelScope, + started = SharingStarted.Eagerly, + initialValue = null to -1, + ) - val selectedFrame = state - .map { state -> - state.selectedFrame() to state.frameIndex - }.stateIn( - scope = viewModelScope, - started = SharingStarted.Eagerly, - initialValue = null to -1, - ) + val allFrames = + selectedThread + .map { (thread, _) -> + thread?.getFrames() ?: emptyList() + }.stateIn( + scope = viewModelScope, + started = SharingStarted.Eagerly, + initialValue = emptyList(), + ) - val selectedFrameVariables = selectedFrame - .map { (frame, _) -> - frame?.getVariables() ?: emptyList() - }.stateIn( - scope = viewModelScope, - started = SharingStarted.Eagerly, - initialValue = emptyList(), - ) + val selectedFrame = + state + .map { state -> + state.selectedFrame() to state.frameIndex + }.stateIn( + scope = viewModelScope, + started = SharingStarted.Eagerly, + initialValue = null to -1, + ) - val variablesTree = state - .map { state -> - state.variablesTree - }.stateIn( - scope = viewModelScope, - started = SharingStarted.Eagerly, - initialValue = DebuggerState.DEFAULT.variablesTree, - ) + val selectedFrameVariables = + selectedFrame + .map { (frame, _) -> + frame?.getVariables() ?: emptyList() + }.stateIn( + scope = viewModelScope, + started = SharingStarted.Eagerly, + initialValue = emptyList(), + ) + + val variablesTree = + state + .map { state -> + state.variablesTree + }.stateIn( + scope = viewModelScope, + started = SharingStarted.Eagerly, + initialValue = DebuggerState.DEFAULT.variablesTree, + ) override fun onCleared() { super.onCleared() Lookup.getDefault().unregister(IDEDebugClientImpl::class.java) + + // The client owns two thread-backed coroutine contexts. Without this, every destroyed + // editor activity leaks those threads for the life of the process (ADFA-5375). + debugClient.close() } fun setConnectionState(state: DebuggerConnectionState) { _connectionState.update { state } } - private fun setDebuggerState(newState: DebuggerState, because: String) { + private fun setDebuggerState( + newState: DebuggerState, + because: String, + ) { logger.debug("Updating debugger state because {}: {}", because, newState) state.update { newState } } private suspend inline fun setDebuggerState( because: String, - crossinline newState: suspend (DebuggerState) -> DebuggerState + crossinline newState: suspend (DebuggerState) -> DebuggerState, ) { val currentState = state.value val newState = newState(currentState) @@ -229,13 +243,13 @@ class DebuggerViewModel : ViewModel() { suspend fun setThreads( threads: List, selectedThreadIndex: Int = -1, - selectedFrameIndex: Int = -1 + selectedFrameIndex: Int = -1, ) { logger.debug( "setThreads(selectedThreadIndex={}, selectedFrameIndex={}, threads={})", selectedThreadIndex, selectedFrameIndex, - threads + threads, ) withContext(Dispatchers.IO) { @@ -270,11 +284,16 @@ class DebuggerViewModel : ViewModel() { var frameIndex = selectedFrameIndex if (frameIndex < 0) { - frameIndex = if (resolvableThreads - .getOrNull(threadIndex) - ?.getFrames() - ?.firstOrNull() != null - ) 0 else -1 + frameIndex = + if (resolvableThreads + .getOrNull(threadIndex) + ?.getFrames() + ?.firstOrNull() != null + ) { + 0 + } else { + -1 + } } DebuggerState( @@ -343,38 +362,39 @@ class DebuggerViewModel : ViewModel() { } } - suspend fun setSelectedThreadIndex(index: Int) = withContext(Dispatchers.IO) { - setDebuggerState(because = "selected thread index changed") { current -> - check(index in 0.. + check(index in 0.. - check(index in 0..<(current.selectedThread?.getFrames()?.size ?: 0)) { - "Invalid frame index: $index" - } + suspend fun setSelectedFrameIndex(index: Int) = + withContext(Dispatchers.IO) { + setDebuggerState(because = "selected frame index changed") { current -> + check(index in 0..<(current.selectedThread?.getFrames()?.size ?: 0)) { + "Invalid frame index: $index" + } - current.copy( - frameIndex = index, - variablesTree = - createVariablesTree( - current.threads, - current.threadIndex, - index, - ), - ) + current.copy( + frameIndex = index, + variablesTree = + createVariablesTree( + current.threads, + current.threadIndex, + index, + ), + ) + } } - } @OptIn(ExperimentalStdlibApi::class) fun observeLatestSelectedFrame( diff --git a/app/src/test/java/com/itsaky/androidide/viewmodel/DebuggerThreadLeakTest.kt b/app/src/test/java/com/itsaky/androidide/viewmodel/DebuggerThreadLeakTest.kt new file mode 100644 index 0000000000..ce6c797f39 --- /dev/null +++ b/app/src/test/java/com/itsaky/androidide/viewmodel/DebuggerThreadLeakTest.kt @@ -0,0 +1,175 @@ +package com.itsaky.androidide.viewmodel + +import androidx.lifecycle.ViewModelProvider +import androidx.lifecycle.ViewModelStore +import com.google.common.truth.Truth.assertThat +import com.itsaky.androidide.app.BaseApplication +import com.itsaky.androidide.lsp.BreakpointHandler +import com.itsaky.androidide.projects.IProjectManager +import com.itsaky.androidide.projects.ProjectManagerImpl +import com.itsaky.androidide.repositories.BreakpointRepository +import kotlinx.coroutines.ExperimentalCoroutinesApi +import org.junit.Before +import org.junit.Rule +import org.junit.Test +import org.junit.rules.TemporaryFolder +import org.junit.runner.RunWith +import org.robolectric.RobolectricTestRunner +import org.robolectric.annotation.Config +import java.io.File + +/** + * ADFA-5375: a debugger session must not cost the process a thread that is never given back. + * + * LeakCanary cannot see this - it watches objects - so the guard is a thread count, the same + * measurement that found the leak on device. + */ +@OptIn(ExperimentalCoroutinesApi::class) +@RunWith(RobolectricTestRunner::class) +@Config(application = DebuggerThreadLeakTest.TestApp::class) +class DebuggerThreadLeakTest { + /** IDEApplication's async loaders call exitProcess when they fail under Robolectric. */ + open class TestApp : BaseApplication() + + @get:Rule + val mainDispatcherRule = MainDispatcherRule() + + @get:Rule + val projectFolder = TemporaryFolder() + + @Before + fun setUp() { + // Breakpoint persistence resolves against the project dir. Without this it resolves against + // the module directory and drops an untracked file into the worktree. + (IProjectManager.getInstance() as ProjectManagerImpl).projectPath = + projectFolder.root.absolutePath + } + + @Test + fun `repeated view model lifecycles do not accumulate threads`() { + // The shared IO dispatcher grows once, on demand, so an absolute count says nothing. What + // must not happen is growth that keeps tracking the number of sessions: seven open/close + // cycles left seven live debugger threads on device before the fix. + runCycles(CYCLES) + val afterWarmup = settledThreadCount() + + runCycles(CYCLES) + + assertThat(settledThreadCount()).isAtMost(afterWarmup + SLACK) + } + + private fun runCycles(count: Int) { + repeat(count) { + val store = ViewModelStore() + try { + newViewModel(store).debugClient.toggleBreakpoint(sourceFile(), 1) + } finally { + store.clear() + } + } + } + + @Test + fun `clearing the view model closes the debug client`() { + val store = ViewModelStore() + val viewModel = newViewModel(store) + try { + assertThat(viewModel.debugClient.isClientScopeActive).isTrue() + } finally { + store.clear() + } + + assertThat(viewModel.debugClient.isClientScopeActive).isFalse() + } + + @Test + fun `a closed client ignores further breakpoint work`() { + val store = ViewModelStore() + val client = newViewModel(store).debugClient + store.clear() + + client.toggleBreakpoint(sourceFile(), 3) + + // The guard must reject the call outright; a cancelled scope would swallow it silently. + assertThat(client.breakpoints.allBreakpoints).isEmpty() + } + + @Test + fun `closing writes a breakpoint made inside the debounce window, without waiting it out`() { + val handler = BreakpointHandler() + val stored = BreakpointRepository.getBreakpointsStorageFile(projectFolder.root) + val closedAt: Long + try { + handler.begin { } + handler.toggle(sourceFile(), BREAKPOINT_LINE) + awaitTrue("breakpoint registered") { handler.allBreakpoints.isNotEmpty() } + } finally { + // Well inside the save debounce. The write must survive the close (it runs on a scope + // close does not cancel) and must not wait out the remaining delay. + closedAt = System.currentTimeMillis() + handler.close() + } + + awaitTrue("breakpoints written to $stored") { stored.exists() } + assertThat(System.currentTimeMillis() - closedAt).isLessThan(SAVE_DEBOUNCE_MS) + assertThat(stored.readText().filterNot(Char::isWhitespace)) + .contains("\"line\":$BREAKPOINT_LINE") + } + + private fun newViewModel(store: ViewModelStore) = + ViewModelProvider(store, ViewModelProvider.NewInstanceFactory()) + .get(DebuggerViewModel::class.java) + + private fun sourceFile() = File(projectFolder.root, "app/src/main/java/Main.java") + + private fun jvmThreadCount(): Int { + var group = Thread.currentThread().threadGroup!! + while (group.parent != null) { + group = group.parent!! + } + return group.activeCount() + } + + /** Lets the just-finished cycles' coroutines drain before sampling. */ + private fun settledThreadCount(): Int { + var last = jvmThreadCount() + val deadline = System.currentTimeMillis() + SETTLE_MS + while (System.currentTimeMillis() < deadline) { + Thread.sleep(POLL_MS) + val now = jvmThreadCount() + if (now == last) { + return now + } + last = now + } + + return last + } + + private fun awaitTrue( + what: String, + condition: () -> Boolean, + ) { + val deadline = System.currentTimeMillis() + TIMEOUT_MS + while (System.currentTimeMillis() < deadline) { + if (condition()) { + return + } + Thread.sleep(POLL_MS) + } + + throw AssertionError("Timed out waiting for: $what") + } + + companion object { + private const val CYCLES = 7 + private const val SLACK = 2 + private const val BREAKPOINT_LINE = 5 + private const val TIMEOUT_MS = 5_000L + private const val SETTLE_MS = 2_000L + + /** Comfortably under BreakpointHandler's 1s debounce, comfortably over a prompt flush. */ + private const val SAVE_DEBOUNCE_MS = 500L + private const val POLL_MS = 25L + } +} diff --git a/lsp/java/src/main/java/com/itsaky/androidide/lsp/java/debug/EventHandler.kt b/lsp/java/src/main/java/com/itsaky/androidide/lsp/java/debug/EventHandler.kt index a129eac02a..38a9e9e9f7 100644 --- a/lsp/java/src/main/java/com/itsaky/androidide/lsp/java/debug/EventHandler.kt +++ b/lsp/java/src/main/java/com/itsaky/androidide/lsp/java/debug/EventHandler.kt @@ -23,12 +23,12 @@ import com.sun.jdi.event.WatchpointEvent import com.sun.jdi.request.EventRequest import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope -import kotlinx.coroutines.DelicateCoroutinesApi +import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.Job +import kotlinx.coroutines.cancel import kotlinx.coroutines.job import kotlinx.coroutines.launch -import kotlinx.coroutines.newSingleThreadContext import org.slf4j.LoggerFactory /** @@ -37,278 +37,281 @@ import org.slf4j.LoggerFactory * @author Akash Yadav */ internal class EventHandler( - private val vm: VirtualMachine, - private val threadState: ThreadState, - private val stopOnVmStart: Boolean, - private val consumer: EventConsumer + private val vm: VirtualMachine, + private val threadState: ThreadState, + private val stopOnVmStart: Boolean, + private val consumer: EventConsumer, ) : AutoCloseable { - - internal val eventRequestSpecList = EventRequestSpecList(vm) - - @Volatile - private var connected = true - private var vmDied = false - private var completed = false - - @OptIn(DelicateCoroutinesApi::class, ExperimentalCoroutinesApi::class) - private val adapterContext = newSingleThreadContext("JDWPEventHandler") - private val adapterScope = CoroutineScope(adapterContext) - private var eventsJob: Job? = null - - companion object { - private val logger = LoggerFactory.getLogger(EventHandler::class.java) - } - - /** - * Run the event handler. - */ - fun startListening() { - eventsJob = adapterScope.launch { - val job = coroutineContext.job - val queue = vm.eventQueue() - while (connected && job.isActive && !job.isCancelled) { - try { - val events = queue.remove() - var resumeVm = false - for (event in events.eventIterator()) { - logger.info("startListening: received event: {}", event) - val stopForEvent = handleEvent(event) - logger.info( - "startListening: handled event: {}, resumeVm={}, stopForEvent={}", - event, - resumeVm, - stopForEvent - ) - - resumeVm = resumeVm || !stopForEvent - } - - if (resumeVm) { - logger.debug("resuming VM") - events.resume() - } else if (events.suspendPolicy() == EventRequest.SUSPEND_ALL) { - // notify consumer that the VM has interrupted - logger.info("startListening: VM interrupted") - setCurrentThread(events) - consumer.vmInterrupted() - } - } catch (interrupt: InterruptedException) { - logger.debug("event handler interrupted") - // Ignore, changes will be seen at top of the loop - } catch (err: VMDisconnectedException) { - handleDisconnectedException() - } - } - - completed = true - eventsJob = null - logger.info("EventHandler completed") - } - } - - private fun setCurrentThread(set: EventSet) { - val thread: ThreadReference? - if (set.size > 0) { - /* - * If any event in the set has a thread associated with it, - * they all will, so just grab the first one. - */ - val event = set.iterator().next() // Is there a better way? - thread = eventThread(event) - } else { - thread = null - } - setCurrentThread(thread) - } - - private fun setCurrentThread(thread: ThreadReference?) { - threadState.invalidateAll() - threadState.setCurrentThread(thread) - } - - /** - * Handle the event. - * - * @param event The event to handle. - * @return `true` if the VM should be stopped, false otherwise. - */ - private fun handleEvent(event: Event): Boolean { - consumer.receivedEvent(event) - - return when (event) { - is ExceptionEvent -> exceptionEvent(event) - is BreakpointEvent -> breakpointEvent(event) - is WatchpointEvent -> fieldWatchEvent(event) - is StepEvent -> stepEvent(event) - is MethodEntryEvent -> methodEntryEvent(event) - is MethodExitEvent -> methodExitEvent(event) - is ClassPrepareEvent -> classPrepareEvent(event) - is ClassUnloadEvent -> classUnloadEvent(event) - is ThreadStartEvent -> threadStartEvent(event) - is ThreadDeathEvent -> threadDeathEvent(event) - is VMStartEvent -> vmStartEvent(event) - else -> handleExitEvent(event) - } - } - - /** - * @see [handleEvent] - */ - private fun vmStartEvent(event: VMStartEvent): Boolean { - consumer.vmStartEvent(event) - return stopOnVmStart - } - - /** - * @see [handleEvent] - */ - private fun breakpointEvent(event: BreakpointEvent): Boolean { - consumer.breakpointEvent(event) - return true - } - - /** - * @see [handleEvent] - */ - private fun methodEntryEvent(event: MethodEntryEvent): Boolean { - consumer.methodEntryEvent(event) - return true - } - - /** - * @see [handleEvent] - */ - private fun methodExitEvent(event: MethodExitEvent): Boolean { - return consumer.methodExitEvent(event) - } - - /** - * @see [handleEvent] - */ - private fun fieldWatchEvent(event: WatchpointEvent): Boolean { - consumer.fieldWatchEvent(event) - return true - } - - /** - * @see [handleEvent] - */ - private fun stepEvent(event: StepEvent): Boolean { - consumer.stepEvent(event) - return true - } - - /** - * @see [handleEvent] - */ - private fun classPrepareEvent(event: ClassPrepareEvent): Boolean { - consumer.classPrepareEvent(event) + internal val eventRequestSpecList = EventRequestSpecList(vm) + + @Volatile + private var connected = true + private var vmDied = false + private var completed = false + + // A handler exists per debug session, so a thread-owning dispatcher here leaked a thread per + // session; limitedParallelism(1) keeps the sequencing and owns nothing (ADFA-5397). + @OptIn(ExperimentalCoroutinesApi::class) + private val adapterContext = Dispatchers.IO.limitedParallelism(1) + private val adapterScope = CoroutineScope(adapterContext) + private var eventsJob: Job? = null + + companion object { + private val logger = LoggerFactory.getLogger(EventHandler::class.java) + } + + /** + * Run the event handler. + */ + fun startListening() { + eventsJob = + adapterScope.launch { + val job = coroutineContext.job + val queue = vm.eventQueue() + while (connected && job.isActive && !job.isCancelled) { + try { + val events = queue.remove() + var resumeVm = false + for (event in events.eventIterator()) { + logger.info("startListening: received event: {}", event) + val stopForEvent = handleEvent(event) + logger.info( + "startListening: handled event: {}, resumeVm={}, stopForEvent={}", + event, + resumeVm, + stopForEvent, + ) + + resumeVm = resumeVm || !stopForEvent + } + + if (resumeVm) { + logger.debug("resuming VM") + events.resume() + } else if (events.suspendPolicy() == EventRequest.SUSPEND_ALL) { + // notify consumer that the VM has interrupted + logger.info("startListening: VM interrupted") + setCurrentThread(events) + consumer.vmInterrupted() + } + } catch (interrupt: InterruptedException) { + logger.debug("event handler interrupted") + // Ignore, changes will be seen at top of the loop + } catch (err: VMDisconnectedException) { + handleDisconnectedException() + } + } + + completed = true + eventsJob = null + logger.info("EventHandler completed") + } + } + + private fun setCurrentThread(set: EventSet) { + val thread: ThreadReference? + if (set.size > 0) { + /* + * If any event in the set has a thread associated with it, + * they all will, so just grab the first one. + */ + val event = set.iterator().next() // Is there a better way? + thread = eventThread(event) + } else { + thread = null + } + setCurrentThread(thread) + } + + private fun setCurrentThread(thread: ThreadReference?) { + threadState.invalidateAll() + threadState.setCurrentThread(thread) + } + + /** + * Handle the event. + * + * @param event The event to handle. + * @return `true` if the VM should be stopped, false otherwise. + */ + private fun handleEvent(event: Event): Boolean { + consumer.receivedEvent(event) + + return when (event) { + is ExceptionEvent -> exceptionEvent(event) + is BreakpointEvent -> breakpointEvent(event) + is WatchpointEvent -> fieldWatchEvent(event) + is StepEvent -> stepEvent(event) + is MethodEntryEvent -> methodEntryEvent(event) + is MethodExitEvent -> methodExitEvent(event) + is ClassPrepareEvent -> classPrepareEvent(event) + is ClassUnloadEvent -> classUnloadEvent(event) + is ThreadStartEvent -> threadStartEvent(event) + is ThreadDeathEvent -> threadDeathEvent(event) + is VMStartEvent -> vmStartEvent(event) + else -> handleExitEvent(event) + } + } + + /** + * @see [handleEvent] + */ + private fun vmStartEvent(event: VMStartEvent): Boolean { + consumer.vmStartEvent(event) + return stopOnVmStart + } + + /** + * @see [handleEvent] + */ + private fun breakpointEvent(event: BreakpointEvent): Boolean { + consumer.breakpointEvent(event) + return true + } + + /** + * @see [handleEvent] + */ + private fun methodEntryEvent(event: MethodEntryEvent): Boolean { + consumer.methodEntryEvent(event) + return true + } + + /** + * @see [handleEvent] + */ + private fun methodExitEvent(event: MethodExitEvent): Boolean = consumer.methodExitEvent(event) + + /** + * @see [handleEvent] + */ + private fun fieldWatchEvent(event: WatchpointEvent): Boolean { + consumer.fieldWatchEvent(event) + return true + } + + /** + * @see [handleEvent] + */ + private fun stepEvent(event: StepEvent): Boolean { + consumer.stepEvent(event) + return true + } + + /** + * @see [handleEvent] + */ + private fun classPrepareEvent(event: ClassPrepareEvent): Boolean { + consumer.classPrepareEvent(event) val success = eventRequestSpecList.resolve(event) if (!success) { logger.error("Error resolving event request for event: $event") } return false - } - - - /** - * @see [handleEvent] - */ - private fun classUnloadEvent(event: ClassUnloadEvent): Boolean { - consumer.classUnloadEvent(event) - return false - } - - /** - * @see [handleEvent] - */ - private fun exceptionEvent(event: ExceptionEvent): Boolean { - consumer.exceptionEvent(event) - return true - } - - /** - * @see [handleEvent] - */ - private fun threadStartEvent(event: ThreadStartEvent): Boolean { - threadState.addThread(event.thread()) - consumer.threadStartEvent(event) - return false - } - - /** - * @see [handleEvent] - */ - private fun threadDeathEvent(event: ThreadDeathEvent): Boolean { - threadState.addThread(event.thread()) - consumer.threadDeathEvent(event) - return false - } - - private fun eventThread(event: Event) = when (event) { - is ClassPrepareEvent -> event.thread() - is LocatableEvent -> event.thread() - is ThreadStartEvent -> event.thread() - is ThreadDeathEvent -> event.thread() - is VMStartEvent -> event.thread() - else -> null - } - - @Synchronized - private fun handleDisconnectedException() { - // Flush the event queue. Dealing only with vm death or disconnection to ensure - // proper termination of this handler - - val queue = vm.eventQueue() - while (connected) { - try { - val eventSet = queue.remove() - val iter = eventSet.eventIterator() - while (iter.hasNext()) { - handleExitEvent(iter.next()) - } - } catch (exc: InterruptedException) { - // ignore - } catch (exc: InternalError) { - // ignore - } - } - } - - private fun handleExitEvent(event: Event): Boolean { - when (event) { - is VMDeathEvent -> { - vmDied = true - return vmDeathEvent(event) - } - - is VMDisconnectEvent -> { - connected = false - if (!vmDied) { - vmDisconnectEvent(event) - } - - return false - } - - else -> throw IllegalArgumentException("Unknown event type: $event") - } - } - - private fun vmDeathEvent(event: VMDeathEvent): Boolean { - consumer.vmDeathEvent(event) - return false - } - - private fun vmDisconnectEvent(event: VMDisconnectEvent): Boolean { - consumer.vmDisconnectEvent(event) - return false - } - - override fun close() { - connected = false - eventsJob?.cancel(CancellationException("EventHandler closed")) - eventsJob = null - } -} \ No newline at end of file + } + + /** + * @see [handleEvent] + */ + private fun classUnloadEvent(event: ClassUnloadEvent): Boolean { + consumer.classUnloadEvent(event) + return false + } + + /** + * @see [handleEvent] + */ + private fun exceptionEvent(event: ExceptionEvent): Boolean { + consumer.exceptionEvent(event) + return true + } + + /** + * @see [handleEvent] + */ + private fun threadStartEvent(event: ThreadStartEvent): Boolean { + threadState.addThread(event.thread()) + consumer.threadStartEvent(event) + return false + } + + /** + * @see [handleEvent] + */ + private fun threadDeathEvent(event: ThreadDeathEvent): Boolean { + threadState.addThread(event.thread()) + consumer.threadDeathEvent(event) + return false + } + + private fun eventThread(event: Event) = + when (event) { + is ClassPrepareEvent -> event.thread() + is LocatableEvent -> event.thread() + is ThreadStartEvent -> event.thread() + is ThreadDeathEvent -> event.thread() + is VMStartEvent -> event.thread() + else -> null + } + + @Synchronized + private fun handleDisconnectedException() { + // Flush the event queue. Dealing only with vm death or disconnection to ensure + // proper termination of this handler + + val queue = vm.eventQueue() + while (connected) { + try { + val eventSet = queue.remove() + val iter = eventSet.eventIterator() + while (iter.hasNext()) { + handleExitEvent(iter.next()) + } + } catch (exc: InterruptedException) { + // ignore + } catch (exc: InternalError) { + // ignore + } + } + } + + private fun handleExitEvent(event: Event): Boolean { + when (event) { + is VMDeathEvent -> { + vmDied = true + return vmDeathEvent(event) + } + + is VMDisconnectEvent -> { + connected = false + if (!vmDied) { + vmDisconnectEvent(event) + } + + return false + } + + else -> { + throw IllegalArgumentException("Unknown event type: $event") + } + } + } + + private fun vmDeathEvent(event: VMDeathEvent): Boolean { + consumer.vmDeathEvent(event) + return false + } + + private fun vmDisconnectEvent(event: VMDisconnectEvent): Boolean { + consumer.vmDisconnectEvent(event) + return false + } + + override fun close() { + connected = false + eventsJob?.cancel(CancellationException("EventHandler closed")) + eventsJob = null + adapterScope.cancel(CancellationException("EventHandler closed")) + } +}