From 9504d09f8621ed0d6f852899bc77375fdb5309f5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sergej=20Ko=C5=A1=C4=8Dejev?= Date: Thu, 5 Nov 2026 22:19:35 +0100 Subject: [PATCH 1/3] perf(daemon): refresh before each request from native file watcher events --- .../cli/LibraryModelRefreshIntegrationTest.kt | 50 ++++++- daemon-mps-plugin/build.gradle.kts | 7 + .../plugin/HeadlessNativeFileWatcher.kt | 84 +++++++++++ .../src/main/resources/META-INF/plugin.xml | 4 + .../mops/daemon/ExternalProjectRefresh.kt | 21 +-- .../mops/daemon/FileWatcherExecutable.kt | 66 +++++++++ .../mops/daemon/FileWatcherFence.kt | 85 ++++++++++++ .../mops/daemon/JetBrainsMpsAccess.kt | 2 +- .../mops/daemon/MopsDaemonCommand.kt | 1 + .../mops/daemon/ProjectRequestGate.kt | 22 ++- .../mops/daemon/WatchedFolderRefresh.kt | 131 ++++++++++++++++++ .../daemon/FileWatcherRefreshSemanticsTest.kt | 80 +++++++++++ .../mops/daemon/ProjectRequestGateTest.kt | 77 ++++++---- .../mops/daemon/SharedMpsEnvironment.kt | 5 + docs/daemon-request-concurrency.md | 10 +- docs/test-running-design.md | 6 +- skills/mops-daemon/SKILL.md | 6 + 17 files changed, 600 insertions(+), 57 deletions(-) create mode 100644 daemon-mps-plugin/src/main/kotlin/com/specificlanguages/mops/daemon/plugin/HeadlessNativeFileWatcher.kt create mode 100644 daemon/src/main/kotlin/com/specificlanguages/mops/daemon/FileWatcherExecutable.kt create mode 100644 daemon/src/main/kotlin/com/specificlanguages/mops/daemon/FileWatcherFence.kt create mode 100644 daemon/src/main/kotlin/com/specificlanguages/mops/daemon/WatchedFolderRefresh.kt create mode 100644 daemon/src/test/kotlin/com/specificlanguages/mops/daemon/FileWatcherRefreshSemanticsTest.kt diff --git a/cli/src/integrationTest/kotlin/com/specificlanguages/mops/cli/LibraryModelRefreshIntegrationTest.kt b/cli/src/integrationTest/kotlin/com/specificlanguages/mops/cli/LibraryModelRefreshIntegrationTest.kt index dea1dd1a..2feee361 100644 --- a/cli/src/integrationTest/kotlin/com/specificlanguages/mops/cli/LibraryModelRefreshIntegrationTest.kt +++ b/cli/src/integrationTest/kotlin/com/specificlanguages/mops/cli/LibraryModelRefreshIntegrationTest.kt @@ -1,16 +1,22 @@ package com.specificlanguages.mops.cli +import com.specificlanguages.mops.protocol.DaemonPaths +import com.specificlanguages.mops.protocol.DaemonWorkspace import org.junit.jupiter.api.io.CleanupMode import org.junit.jupiter.api.io.TempDir import org.junit.jupiter.api.parallel.ResourceLock import java.nio.file.Files import java.nio.file.Path import kotlin.io.path.createDirectories +import kotlin.io.path.isRegularFile +import kotlin.io.path.listDirectoryEntries import kotlin.io.path.pathString import kotlin.io.path.readText import kotlin.io.path.writeText import kotlin.test.Test +import kotlin.test.assertContains import kotlin.test.assertEquals +import kotlin.test.assertFalse @ResourceLock("system-streams") class LibraryModelRefreshIntegrationTest { @@ -19,13 +25,41 @@ class LibraryModelRefreshIntegrationTest { @Test fun `external edits of a library model outside the project folder are visible in the same daemon and after a restart`() { + val project = prepareProject() + val daemonHome = tempDir.resolve("daemon-home").createDirectories() + exerciseEdits(project, daemonHome) { args -> + runCommandLine(project, "--daemon-home", daemonHome.pathString, *javaAndMpsHomeArgs(), *args) + } + if (System.getProperty("os.name").startsWith("Mac")) { + assertContains(daemonLogs(daemonHome), "refreshed external changes using file watcher events") + } + } + + @Test + fun `external edits are visible with the native file watcher turned off`() { + val project = prepareProject() + val daemonHome = tempDir.resolve("daemon-home").createDirectories() + val cli = InstalledCli( + project, daemonHome, tempDir, + mapOf("MOPS_DAEMON_JAVA_OPTIONS" to "-Xmx2g -Didea.filewatcher.disabled=true"), + ) + exerciseEdits(project, daemonHome) { args -> cli.run(*args) } + val logs = daemonLogs(daemonHome) + assertContains(logs, "using a full walk because the file watcher is not operational") + assertFalse("using file watcher events" in logs, logs) + } + + private fun prepareProject(): Path { val project = copyTestProject("project-library", tempDir.resolve("project")) - val shared = Files.move(project.resolve("libs"), tempDir.resolve("shared")) + Files.move(project.resolve("libs"), tempDir.resolve("shared")) val libraries = project.resolve(".mps/libraries.xml") libraries.writeText(libraries.readText().replace("${'$'}PROJECT_DIR${'$'}/libs", "${'$'}PROJECT_DIR${'$'}/../shared")) - val libraryModel = shared.resolve("library/models/values.mps") + return project + } + + private fun exerciseEdits(project: Path, daemonHome: Path, run: (Array) -> CliResult) { + val libraryModel = tempDir.resolve("shared/library/models/values.mps") val projectModel = project.resolve("solutions/tests/models/library@tests.mps") - val daemonHome = tempDir.resolve("daemon-home").createDirectories() val program = tempDir.resolve("names.groovy").also { it.writeText(""" project.read { @@ -35,10 +69,8 @@ class LibraryModelRefreshIntegrationTest { } """.trimIndent()) } - fun run(vararg args: String): CliResult = - runCommandLine(project, "--daemon-home", daemonHome.pathString, *javaAndMpsHomeArgs(), *args) fun names(): String { - val result = run("code", "run", program.pathString) + val result = run(arrayOf("code", "run", program.pathString)) assertEquals(0, result.exitCode, result.output) return result.stdout.trim() } @@ -58,11 +90,15 @@ class LibraryModelRefreshIntegrationTest { rename(libraryModel, "LibraryValue", "LibraryValueEdited") assertEquals("LibraryValueEdited;LibraryPassEdited", names(), "library model edit in the same daemon") - assertEquals(0, run("daemon", "stop").exitCode) + assertEquals(0, run(arrayOf("daemon", "stop")).exitCode) rename(libraryModel, "LibraryValueEdited", "LibraryValueAfterRestart") assertEquals("LibraryValueAfterRestart;LibraryPassEdited", names(), "edit while the daemon was stopped") } finally { stopDaemons(project, daemonHome) } } + + private fun daemonLogs(daemonHome: Path): String = + DaemonPaths(daemonHome).projects.listDirectoryEntries() + .map { DaemonWorkspace(it).logFile() }.filter { it.isRegularFile() }.joinToString("\n") { it.readText() } } diff --git a/daemon-mps-plugin/build.gradle.kts b/daemon-mps-plugin/build.gradle.kts index 9185c202..65708aac 100644 --- a/daemon-mps-plugin/build.gradle.kts +++ b/daemon-mps-plugin/build.gradle.kts @@ -18,6 +18,13 @@ dependencies { include("lib/mps-openapi.jar") include("lib/mps-persistence.jar") include("lib/util.jar") + // NativeFileWatcherImpl, PluggableFileWatcher, Notifications and their supertypes, for + // HeadlessNativeFileWatcher. + include("lib/util-8.jar") + include("lib/app.jar") + include("lib/intellij.platform.core.jar") + include("lib/intellij.platform.ide.core.jar") + include("lib/intellij.platform.ide.impl.jar") include("plugins/mps-java/lib/java-core.jar") include("plugins/java/lib/ecj/eclipse.jar") } diff --git a/daemon-mps-plugin/src/main/kotlin/com/specificlanguages/mops/daemon/plugin/HeadlessNativeFileWatcher.kt b/daemon-mps-plugin/src/main/kotlin/com/specificlanguages/mops/daemon/plugin/HeadlessNativeFileWatcher.kt new file mode 100644 index 00000000..ac0b53e7 --- /dev/null +++ b/daemon-mps-plugin/src/main/kotlin/com/specificlanguages/mops/daemon/plugin/HeadlessNativeFileWatcher.kt @@ -0,0 +1,84 @@ +package com.specificlanguages.mops.daemon.plugin + +import com.intellij.notification.Notification +import com.intellij.notification.Notifications +import com.intellij.openapi.application.ApplicationManager +import com.intellij.openapi.vfs.impl.local.NativeFileWatcherImpl +import java.util.concurrent.atomic.AtomicInteger +import java.util.concurrent.atomic.AtomicLong + +/** + * The platform's native file watcher (`fsnotifier`), enabled in a headless application. The platform registers its own + * instance, which stays off in command-line and test applications; this second registration differs only in that + * check, and can still be turned off with `-Didea.filewatcher.disabled=true`. + * + * The daemon reads [changeCount] and [recursiveRoots] by reflection, because this class is loaded by the plugin class + * loader. + */ +class HeadlessNativeFileWatcher : NativeFileWatcherImpl() { + private val rootChanges = AtomicLong() + + @Volatile + private var recursive: List = emptyList() + + @Volatile + private var flat: List = emptyList() + + @Volatile + private var failed = false + + private val startAttempts: AtomicInteger? = startAttemptCounter() + + init { + ApplicationManager.getApplication().messageBus.connect().subscribe( + Notifications.TOPIC, + object : Notifications { + override fun notify(notification: Notification) { + if (notification.groupId == FAILURE_GROUP) failed = true + } + }, + ) + } + + override fun isDisabled(): Boolean = java.lang.Boolean.getBoolean(DISABLED_PROPERTY) + + override fun setWatchRoots(recursive: List, flat: List, shuttingDown: Boolean) { + if (recursive != this.recursive || flat != this.flat) { + this.recursive = recursive + this.flat = flat + rootChanges.incrementAndGet() + } + super.setWatchRoots(recursive, flat, shuttingDown) + } + + /** + * A number that changes whenever the watcher may have missed changes: when the watch roots change and when the + * `fsnotifier` process is started again. `null` once the watcher has reported a failure, such as the Linux inotify + * watch limit, or when the start count cannot be read. + */ + fun changeCount(): Long? { + val starts = startAttempts ?: return null + return if (failed) null else rootChanges.get() + starts.get() + } + + /** The recursive watch roots as last set, canonical and system-dependent. */ + fun recursiveRoots(): List = recursive + + /** + * The platform restarts a terminated `fsnotifier` without reporting anything as changed, but counts each start in a + * private field. + */ + private fun startAttemptCounter(): AtomicInteger? = + try { + NativeFileWatcherImpl::class.java.getDeclaredField("myStartAttemptCount") + .apply { isAccessible = true } + .get(this) as? AtomicInteger + } catch (_: ReflectiveOperationException) { + null + } + + private companion object { + const val DISABLED_PROPERTY = "idea.filewatcher.disabled" + const val FAILURE_GROUP = "File Watcher Messages" + } +} diff --git a/daemon-mps-plugin/src/main/resources/META-INF/plugin.xml b/daemon-mps-plugin/src/main/resources/META-INF/plugin.xml index cd0ff020..032bf68b 100644 --- a/daemon-mps-plugin/src/main/resources/META-INF/plugin.xml +++ b/daemon-mps-plugin/src/main/resources/META-INF/plugin.xml @@ -9,4 +9,8 @@ + + + diff --git a/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/ExternalProjectRefresh.kt b/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/ExternalProjectRefresh.kt index 728d6c9b..93704b41 100644 --- a/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/ExternalProjectRefresh.kt +++ b/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/ExternalProjectRefresh.kt @@ -3,7 +3,6 @@ package com.specificlanguages.mops.daemon import com.intellij.openapi.application.ApplicationManager import com.intellij.configurationStore.StoreReloadManager import com.intellij.openapi.vfs.LocalFileSystem -import com.intellij.openapi.vfs.VfsUtil import jetbrains.mps.classloading.ClassLoaderManager import jetbrains.mps.ide.platform.watching.ReloadManager import jetbrains.mps.project.MPSProject @@ -11,7 +10,6 @@ import jetbrains.mps.project.Project import jetbrains.mps.project.facets.JavaModuleFacet import jetbrains.mps.project.dependency.GlobalModuleDependenciesManager import jetbrains.mps.util.PathManager -import jetbrains.mps.vfs.refresh.DefaultCachingContext import kotlinx.coroutines.runBlocking import org.jetbrains.mps.openapi.module.SModuleReference import java.nio.file.Files @@ -19,25 +17,28 @@ import java.nio.file.Path /** * Refreshes project files, the folders of source modules, and the archives of packaged modules, and replaces runtimes - * whose Java classpath contents changed between requests. + * whose Java classpath contents changed between requests. [WatchedFolderRefresh] decides how folders are refreshed. */ -internal class ExternalProjectRefresh(private val project: Project) { +internal class ExternalProjectRefresh(private val project: Project, private val logger: DaemonLogger) { + private val folderRefresh = WatchedFolderRefresh() private val classPathFingerprints = ClassPathFingerprints() private var outputs = snapshot() private var archiveStamps: Map? = null private val foldersWithStoredFilesLoaded = mutableSetOf() private val mpsHome = Path.of(PathManager.getHomePath()) - fun refresh() { + /** Refreshes external changes and returns how the folders were refreshed. */ + fun refresh(): WatchedFolderRefresh.Pass { val mpsProject = project as MPSProject + val pass = folderRefresh.begin() ApplicationManager.getApplication().invokeAndWait { val settings = Path.of(requireNotNull(mpsProject.project.basePath)).resolve(".mps") - mpsProject.fileSystem.getFile(settings.toString()).refresh(DefaultCachingContext(true, true)) + LocalFileSystem.getInstance().refreshAndFindFileByNioFile(settings)?.let { pass.refresh(listOf(it)) } } // The component store reloads project membership separately from MPS's model/module reload sessions. runBlocking { StoreReloadManager.getInstance(mpsProject.project).reloadChangedStorageFiles() } ApplicationManager.getApplication().invokeAndWait { - refreshModuleFolders() + refreshModuleFolders(pass) refreshModuleArchives() ReloadManager.getInstance().flush() val current = snapshot() @@ -50,6 +51,8 @@ internal class ExternalProjectRefresh(private val project: Project) { } outputs = current } + logger.log(pass.summary()) + return pass } /** @@ -58,11 +61,11 @@ internal class ExternalProjectRefresh(private val project: Project) { * folder is seen, the files the VFS has stored under it are loaded into memory before the refresh; * docs/test-running-design.md explains why. */ - private fun refreshModuleFolders() { + private fun refreshModuleFolders(pass: WatchedFolderRefresh.Pass) { val folders = project.modelAccess.computeReadAction { moduleFolders(project, mpsHome) } .mapNotNull { LocalFileSystem.getInstance().findFileByNioFile(it) } folders.filter { foldersWithStoredFilesLoaded.add(it.toNioPath()) }.forEach(::loadStoredFiles) - VfsUtil.markDirtyAndRefresh(false, true, true, *folders.toTypedArray()) + pass.refresh(folders) } /** diff --git a/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/FileWatcherExecutable.kt b/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/FileWatcherExecutable.kt new file mode 100644 index 00000000..891e6df7 --- /dev/null +++ b/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/FileWatcherExecutable.kt @@ -0,0 +1,66 @@ +package com.specificlanguages.mops.daemon + +import java.io.IOException +import java.nio.file.Files +import java.nio.file.Path +import java.nio.file.StandardCopyOption +import kotlin.io.path.createDirectories +import kotlin.io.path.isRegularFile + +/** + * Prepares the native file watcher's executable (`fsnotifier`) before MPS starts. The generic MPS distribution archive + * does not keep the executable bit, so the distribution's binary is copied into a directory of the daemon, made + * executable, and passed to the platform through `idea.filewatcher.executable.path`. Without a binary for this + * platform the watcher is turned off through `idea.filewatcher.disabled`, and refreshes walk folders instead. + * + * Does nothing when either property is already set, so `-Didea.filewatcher.disabled=true` in the daemon's Java options + * turns the watcher off. + */ +internal object FileWatcherExecutable { + private const val EXECUTABLE_PROPERTY = "idea.filewatcher.executable.path" + private const val DISABLED_PROPERTY = "idea.filewatcher.disabled" + + fun install(mpsHome: Path, directory: Path, logger: DaemonLogger) { + if (System.getProperty(EXECUTABLE_PROPERTY) != null || System.getProperty(DISABLED_PROPERTY) != null) return + val bundled = bundled(mpsHome) + if (bundled == null) { + logger.log("no native file watcher in $mpsHome for this platform; refreshes walk folders") + System.setProperty(DISABLED_PROPERTY, "true") + return + } + try { + System.setProperty(EXECUTABLE_PROPERTY, copy(bundled, directory).toString()) + } catch (exception: IOException) { + logger.log("cannot prepare the native file watcher $bundled: $exception; refreshes walk folders") + System.setProperty(DISABLED_PROPERTY, "true") + } + } + + private fun copy(bundled: Path, directory: Path): Path { + val target = directory.createDirectories().resolve(bundled.fileName) + if (!target.isRegularFile() || Files.mismatch(bundled, target) != -1L) { + val temporary = Files.createTempFile(directory, bundled.fileName.toString(), ".tmp") + Files.copy(bundled, temporary, StandardCopyOption.REPLACE_EXISTING) + Files.move(temporary, target, StandardCopyOption.REPLACE_EXISTING, StandardCopyOption.ATOMIC_MOVE) + } + if (!Files.isExecutable(target) && !target.toFile().setExecutable(true)) { + throw IOException("cannot make $target executable") + } + return target + } + + /** The distribution's `fsnotifier` for the current operating system and architecture, if it has one. */ + private fun bundled(mpsHome: Path): Path? { + val os = System.getProperty("os.name") + val arm = System.getProperty("os.arch") in setOf("aarch64", "arm64") + val relative = when { + os.startsWith("Mac") -> "bin/mac/fsnotifier" + os.startsWith("Windows") -> "bin/win/fsnotifier.exe" + // Before MPS 2026.3 the distribution has only an x86-64 binary for Linux. + os.startsWith("Linux") && arm -> "bin/linux-aarch64/fsnotifier" + os.startsWith("Linux") -> "bin/linux/fsnotifier" + else -> return null + } + return mpsHome.resolve(relative).takeIf { it.isRegularFile() } + } +} diff --git a/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/FileWatcherFence.kt b/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/FileWatcherFence.kt new file mode 100644 index 00000000..6b69e0b6 --- /dev/null +++ b/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/FileWatcherFence.kt @@ -0,0 +1,85 @@ +package com.specificlanguages.mops.daemon + +import com.intellij.openapi.application.PathManager +import com.intellij.openapi.util.io.FileUtil +import com.intellij.openapi.vfs.LocalFileSystem +import com.intellij.openapi.vfs.impl.local.FileWatcher +import com.intellij.openapi.vfs.impl.local.LocalFileSystemImpl +import java.io.File +import java.io.IOException +import java.nio.file.Files +import java.nio.file.Path +import java.time.Duration +import java.util.UUID +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import java.util.function.Consumer +import kotlin.io.path.createDirectories +import kotlin.io.path.deleteIfExists + +/** + * Makes sure the native file watcher has reported every change made before [pass] was called. There is no API to + * flush pending watcher events, and on macOS an event arrives up to about 300 ms after the change. `fsnotifier` reports + * events in order, so once it reports a change made by [pass], it has reported every earlier change too. + * + * Each pass creates a fresh directory with a file in it under [root], a folder that is watched recursively, and waits + * until the watcher reports a path inside that directory. A fresh directory can only be named by events that happened + * after it was created. The raw watcher events are observed through `FileWatcher.myTestNotifier`, a private hook that + * the platform calls for every event; see the mps-api-research note native-file-watcher-in-headless-mps.md. + */ +internal class FileWatcherFence private constructor(private val root: Path) { + private val pending = ConcurrentHashMap() + + /** Returns `true` once the watcher reported the fence, or `false` after [timeout]. */ + fun pass(timeout: Duration): Boolean { + val marker = root.resolve(UUID.randomUUID().toString()) + val reported = CountDownLatch(1) + pending[marker.toString()] = reported + try { + Files.createDirectory(marker) + Files.createFile(marker.resolve("fence")) + return reported.await(timeout.toMillis(), TimeUnit.MILLISECONDS) + } finally { + pending.remove(marker.toString()) + deleteQuietly(marker) + } + } + + private fun onEvent(path: String) { + pending.forEach { (marker, reported) -> + if (path == marker || path.startsWith(marker + File.separator)) reported.countDown() + } + } + + private fun deleteQuietly(marker: Path) { + try { + marker.resolve("fence").deleteIfExists() + marker.deleteIfExists() + } catch (_: IOException) { + // A leftover fence directory is harmless: every pass uses a new one. + } + } + + companion object { + /** The fence of this JVM, or `null` when the watcher's event hook is not available. */ + val instance: FileWatcherFence? by lazy { create() } + + private fun create(): FileWatcherFence? { + val fileWatcher = (LocalFileSystem.getInstance() as? LocalFileSystemImpl)?.fileWatcher ?: return null + val hook = try { + FileWatcher::class.java.getDeclaredField("myTestNotifier").apply { isAccessible = true } + } catch (_: ReflectiveOperationException) { + return null + } + if (hook.get(fileWatcher) != null) return null + // Reported paths are canonical and system-dependent, so the root is too. + val root = Path.of(PathManager.getSystemPath(), "mops-file-watcher-fence").createDirectories().toRealPath() + root.toFile().listFiles()?.forEach { FileUtil.delete(it) } + LocalFileSystem.getInstance().addRootToWatch(FileUtil.toSystemIndependentName(root.toString()), true) + val fence = FileWatcherFence(root) + hook.set(fileWatcher, Consumer(fence::onEvent)) + return fence + } + } +} diff --git a/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/JetBrainsMpsAccess.kt b/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/JetBrainsMpsAccess.kt index d6a8ab62..473eaed9 100644 --- a/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/JetBrainsMpsAccess.kt +++ b/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/JetBrainsMpsAccess.kt @@ -37,7 +37,7 @@ class JetBrainsMpsAccess( private val persistence: PersistenceFacade = PersistenceFacade.getInstance(), private val writeTransaction: WriteTransaction = WriteTransaction(), ) : MpsAccess { - private val externalProjectRefresh = ExternalProjectRefresh(project) + private val externalProjectRefresh = ExternalProjectRefresh(project, logger) private val navigationTargets = NavigationTargetResolver(project, persistence) // An MpsRequestException is an expected client error, not a defect. Capture it inside the model action and rethrow diff --git a/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/MopsDaemonCommand.kt b/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/MopsDaemonCommand.kt index c0c76d0a..8e19705b 100644 --- a/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/MopsDaemonCommand.kt +++ b/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/MopsDaemonCommand.kt @@ -63,6 +63,7 @@ class MopsDaemonCommand : Runnable { val workspacePath = Path.of(workspacePath) val mpsPlugin = Path.of(mpsPlugin) + FileWatcherExecutable.install(mpsHome, workspacePath.resolve("fsnotifier"), logger) DaemonRunner( projectPath = projectPath, diff --git a/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/ProjectRequestGate.kt b/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/ProjectRequestGate.kt index a9c7d332..c116bdfc 100644 --- a/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/ProjectRequestGate.kt +++ b/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/ProjectRequestGate.kt @@ -3,21 +3,27 @@ package com.specificlanguages.mops.daemon import java.util.concurrent.locks.ReentrantLock import kotlin.concurrent.withLock -/** FIFO admission outside MPS model actions, excluding refreshes and exclusive work from active readers. */ +/** + * FIFO admission outside MPS model actions, excluding refreshes and exclusive work from active readers. Only requests + * queued before a refresh starts are admitted after it, so every request sees the external changes made before it + * arrived. + */ internal class ProjectRequestGate(private val refresh: () -> Unit) { private val lock = ReentrantLock() private val changed = lock.newCondition() private val waiting = ArrayDeque() private var active = 0 private var refreshing = false + private var queued = 0L fun run(parallelRead: Boolean, operation: () -> T): T { - val ticket = Ticket(parallelRead) - lock.withLock { + val ticket = lock.withLock { + val ticket = Ticket(parallelRead, ++queued) waiting.addLast(ticket) while (!ticket.admitted) { if (active == 0 && !refreshing && waiting.first() === ticket) { refreshing = true + val queuedBeforeRefresh = queued lock.unlock() val failure = try { refresh() @@ -27,19 +33,21 @@ internal class ProjectRequestGate(private val refresh: () -> Unit) { } finally { lock.lock() } - // Readers queued during refresh share its result. The first exclusive ticket ends the batch. + // Readers queued before the refresh started share its result. The first exclusive ticket and the + // first ticket queued during the refresh end the batch. do { val next = waiting.removeFirst() next.failure = failure next.admitted = true active++ - } while (parallelRead && waiting.firstOrNull()?.parallelRead == true) + } while (parallelRead && waiting.firstOrNull()?.sharesRefresh(queuedBeforeRefresh) == true) refreshing = false changed.signalAll() } else { changed.awaitUninterruptibly() } } + ticket } try { ticket.failure?.let { throw it } @@ -52,7 +60,9 @@ internal class ProjectRequestGate(private val refresh: () -> Unit) { } } - private class Ticket(val parallelRead: Boolean) { + private class Ticket(val parallelRead: Boolean, val number: Long) { + fun sharesRefresh(queuedBeforeRefresh: Long) = parallelRead && number <= queuedBeforeRefresh + var admitted = false var failure: Throwable? = null } diff --git a/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/WatchedFolderRefresh.kt b/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/WatchedFolderRefresh.kt new file mode 100644 index 00000000..741aea3d --- /dev/null +++ b/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/WatchedFolderRefresh.kt @@ -0,0 +1,131 @@ +package com.specificlanguages.mops.daemon + +import com.intellij.openapi.application.ApplicationManager +import com.intellij.openapi.vfs.LocalFileSystem +import com.intellij.openapi.vfs.VfsUtil +import com.intellij.openapi.vfs.VirtualFile +import com.intellij.openapi.vfs.impl.local.LocalFileSystemImpl +import com.intellij.openapi.vfs.local.PluggableFileWatcher +import java.io.IOException +import java.nio.file.Path +import java.time.Duration + +/** + * Refreshes local folders before a request, relying on the native file watcher when it can be trusted. + * + * A forced refresh (a full walk) visits every file the VFS has loaded under a folder. With the watcher, a fence first + * makes sure the watcher has reported every change made before the request, and a non-forced recursive refresh then + * visits only the files the watcher reported. The full walk stays the fallback: when the watcher is not operational or + * has reported a failure, when the fence times out, and on the first refresh after the watch roots changed or + * `fsnotifier` was started again. A folder is also walked fully the first time it is refreshed and whenever it lies + * outside every recursive watch root. + * + * See the mps-api-research note native-file-watcher-in-headless-mps.md. + */ +internal class WatchedFolderRefresh { + private var lastChangeCount: Long? = null + private val refreshedFolders = mutableSetOf() + + /** Prepares the refresh for one request. Must run outside the EDT, before [Pass.refresh]. */ + fun begin(): Pass { + val started = System.nanoTime() + val watcher = HeadlessWatcher.find() + val fileWatcher = (LocalFileSystem.getInstance() as LocalFileSystemImpl).fileWatcher + val reason = when { + watcher == null || !fileWatcher.isOperational -> "the file watcher is not operational" + !awaitSettledRoots(fileWatcher::isSettingRoots) -> "the file watcher is still setting its roots" + else -> fenceAndCompare(watcher) + } + return Pass(reason, watcher?.recursiveRoots().orEmpty(), Duration.ofNanos(System.nanoTime() - started)) + } + + private fun fenceAndCompare(watcher: HeadlessWatcher): String? { + val fence = FileWatcherFence.instance + val fenced = fence?.pass(FENCE_TIMEOUT) ?: false + // The watcher reports failures as notifications that are delivered on the EDT. + ApplicationManager.getApplication().invokeAndWait {} + val changeCount = watcher.changeCount() + val previous = lastChangeCount + lastChangeCount = changeCount + return when { + changeCount == null -> "the file watcher reported a failure" + fence == null -> "the file watcher event hook is not available" + !fenced -> "the file watcher did not report the fence within ${FENCE_TIMEOUT.toMillis()} ms" + previous == null -> "this is the first refresh" + previous != changeCount -> "the watch roots changed or the file watcher restarted" + else -> null + } + } + + private fun awaitSettledRoots(settingRoots: () -> Boolean): Boolean { + val deadline = System.nanoTime() + ROOTS_TIMEOUT.toNanos() + while (settingRoots()) { + if (System.nanoTime() > deadline) return false + Thread.sleep(POLL_MILLIS) + } + return true + } + + /** + * The refresh of one request. [fullWalkReason] is `null` when the watcher is trusted and says otherwise why every + * folder is walked. + */ + inner class Pass(val fullWalkReason: String?, private val recursiveRoots: List, private val preparation: Duration) { + private var walked = 0 + private var watched = 0 + private var refreshing = Duration.ZERO + + /** Refreshes [folders] recursively. Must run on the EDT. */ + fun refresh(folders: Collection) { + val started = System.nanoTime() + val (watchedFolders, walkedFolders) = folders.partition { folder -> + val path = folder.toNioPath() + fullWalkReason == null && path in refreshedFolders && isWatched(path) + } + refreshedFolders += folders.map { it.toNioPath() } + if (walkedFolders.isNotEmpty()) VfsUtil.markDirtyAndRefresh(false, true, true, *walkedFolders.toTypedArray()) + if (watchedFolders.isNotEmpty()) LocalFileSystem.getInstance().refreshFiles(watchedFolders, false, true, null) + walked += walkedFolders.size + watched += watchedFolders.size + refreshing += Duration.ofNanos(System.nanoTime() - started) + } + + private fun isWatched(folder: Path): Boolean { + val real = try { + folder.toRealPath() + } catch (_: IOException) { + return false + } + return recursiveRoots.any(real::startsWith) + } + + /** A line for the daemon log that names how the folders were refreshed and what it cost. */ + fun summary(): String { + val mode = if (fullWalkReason == null) "file watcher events" else "a full walk because $fullWalkReason" + return "refreshed external changes using $mode: $watched folders from watcher events, $walked walked; " + + "watcher wait ${preparation.toMillis()} ms, refresh ${refreshing.toMillis()} ms" + } + } + + /** The plugin's `HeadlessNativeFileWatcher`, which is loaded by the plugin class loader and read by reflection. */ + private class HeadlessWatcher(private val instance: PluggableFileWatcher) { + fun changeCount(): Long? = instance.javaClass.getMethod("changeCount").invoke(instance) as Long? + + fun recursiveRoots(): List = + (instance.javaClass.getMethod("recursiveRoots").invoke(instance) as List<*>).map { Path.of(it as String) } + + companion object { + private const val CLASS_NAME = "com.specificlanguages.mops.daemon.plugin.HeadlessNativeFileWatcher" + + fun find(): HeadlessWatcher? = + PluggableFileWatcher.EP_NAME.extensionList.firstOrNull { it.javaClass.name == CLASS_NAME } + ?.let(::HeadlessWatcher) + } + } + + private companion object { + val FENCE_TIMEOUT: Duration = Duration.ofSeconds(2) + val ROOTS_TIMEOUT: Duration = Duration.ofSeconds(10) + const val POLL_MILLIS = 10L + } +} diff --git a/daemon/src/test/kotlin/com/specificlanguages/mops/daemon/FileWatcherRefreshSemanticsTest.kt b/daemon/src/test/kotlin/com/specificlanguages/mops/daemon/FileWatcherRefreshSemanticsTest.kt new file mode 100644 index 00000000..292d49b2 --- /dev/null +++ b/daemon/src/test/kotlin/com/specificlanguages/mops/daemon/FileWatcherRefreshSemanticsTest.kt @@ -0,0 +1,80 @@ +package com.specificlanguages.mops.daemon + +import com.specificlanguages.mops.daemon.core.MpsRead +import com.specificlanguages.mops.protocol.MpsNodeJson +import com.specificlanguages.mops.protocol.NodeTarget +import jetbrains.mps.project.MPSProject +import java.nio.file.Files +import java.nio.file.Path +import kotlin.io.path.readText +import kotlin.io.path.writeText +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNotNull +import kotlin.test.assertNull + +/** + * A refresh sees an external edit made immediately before it, also when it relies on the native file watcher instead + * of walking the folders. On macOS the watcher must be in use, so these tests exercise the fence. + */ +class FileWatcherRefreshSemanticsTest { + + @Test + fun `refresh right after an external edit sees the edit of a project model`() { + SharedMpsEnvironment.withOpenProjectCopy { project, projectPath -> + exerciseEdits(project, projectPath.resolve(STRUCTURE_MODEL_PATH), "JsonFile") { + getNode(NodeTarget.InModel(STRUCTURE_MODEL_NAME, JSON_FILE_NODE_ID)) + } + } + } + + @Test + fun `refresh right after an external edit sees the edit of a library model outside the project folder`() { + lateinit var library: Path + SharedMpsEnvironment.withOpenProjectCopy(LIBRARY_PROJECT, prepare = { project -> + library = Files.move(project.resolve("libs"), project.resolveSibling("shared")) + val libraries = project.resolve(".mps/libraries.xml") + libraries.writeText(libraries.readText().replace("${'$'}PROJECT_DIR${'$'}/libs", "${'$'}PROJECT_DIR${'$'}/../shared")) + }) { project, _ -> + exerciseEdits(project, library.resolve("library/models/values.mps"), "LibraryValue") { + getNode(NodeTarget.InModel(LIBRARY_MODEL_NAME, LIBRARY_ROOT_NODE_ID)) + } + } + } + + private fun exerciseEdits( + project: MPSProject, + model: Path, + originalName: String, + node: MpsRead.() -> MpsNodeJson, + ) { + val access = SharedMpsEnvironment.mpsAccess(project) + val refresh = ExternalProjectRefresh(project, DaemonLogger()) + assertNotNull(refresh.refresh().fullWalkReason, "the first refresh walks every folder") + var previousName = originalName + repeat(EDITS) { index -> + val name = "$originalName$index" + model.writeText(model.readText().replace("value=\"$previousName\"", "value=\"$name\"")) + previousName = name + + val report = refresh.refresh() + + if (isMacOs) assertNull(report.fullWalkReason, "edit $index: the refresh relies on the file watcher") + assertEquals(name, propertyValue(access.read(node), "name"), "edit $index is visible after the refresh") + } + } + + private companion object { + const val EDITS = 20 + val isMacOs = System.getProperty("os.name").startsWith("Mac") + + const val STRUCTURE_MODEL_NAME = "com.specificlanguages.json.structure" + const val STRUCTURE_MODEL_PATH = + "languages/com.specificlanguages.json/models/com.specificlanguages.json.structure.mps" + const val JSON_FILE_NODE_ID = "2110045694544566904" + + const val LIBRARY_PROJECT = "project-library" + const val LIBRARY_MODEL_NAME = "mops.library.values" + const val LIBRARY_ROOT_NODE_ID = "4096" + } +} diff --git a/daemon/src/test/kotlin/com/specificlanguages/mops/daemon/ProjectRequestGateTest.kt b/daemon/src/test/kotlin/com/specificlanguages/mops/daemon/ProjectRequestGateTest.kt index 5564349f..efe469a4 100644 --- a/daemon/src/test/kotlin/com/specificlanguages/mops/daemon/ProjectRequestGateTest.kt +++ b/daemon/src/test/kotlin/com/specificlanguages/mops/daemon/ProjectRequestGateTest.kt @@ -9,50 +9,72 @@ import kotlin.test.* class ProjectRequestGateTest { @Test - fun `read batch shares refresh and exclusive work separates later readers`() { - val refreshing = CountDownLatch(1) - val finishRefresh = CountDownLatch(1) + fun `readers queued before a refresh share it and exclusive work separates later readers`() { + val exclusiveRunning = CountDownLatch(1) + val finishExclusive = CountDownLatch(1) val readers = CountDownLatch(2) val finishReads = CountDownLatch(1) val writing = CountDownLatch(1) val finishWrite = CountDownLatch(1) val refreshes = AtomicInteger() - val gate = ProjectRequestGate { - if (refreshes.incrementAndGet() == 1) { - refreshing.countDown() - finishRefresh.awaitChecked() - } - } - val first = task { gate.run(true) { readers.countDown(); finishReads.awaitChecked() } } + val gate = ProjectRequestGate { refreshes.incrementAndGet() } + val running = task { gate.run(false) { exclusiveRunning.countDown(); finishExclusive.awaitChecked() } } try { - refreshing.awaitChecked() + exclusiveRunning.awaitChecked() + val first = task { gate.run(true) { readers.countDown(); finishReads.awaitChecked() } } + first.awaitQueued() val second = task { gate.run(true) { readers.countDown(); finishReads.awaitChecked() } } second.awaitQueued() val exclusive = task { gate.run(false) { writing.countDown(); finishWrite.awaitChecked() } } exclusive.awaitQueued() val later = task { gate.run(true) { refreshes.get() } } later.awaitQueued() - finishRefresh.countDown() + finishExclusive.countDown() readers.awaitChecked() - assertEquals(1, refreshes.get()) + assertEquals(2, refreshes.get()) assertEquals(1, writing.count) assertFalse(later.future.isDone) finishReads.countDown() writing.awaitChecked() - assertEquals(2, refreshes.get()) + assertEquals(3, refreshes.get()) assertFalse(later.future.isDone) finishWrite.countDown() - assertEquals(3, later.result()) + assertEquals(4, later.result()) + running.result() first.result() second.result() exclusive.result() } finally { - finishRefresh.countDown() + finishExclusive.countDown() finishReads.countDown() finishWrite.countDown() } } + @Test + fun `a reader arriving during a refresh waits for its own refresh`() { + val refreshing = CountDownLatch(1) + val finishRefresh = CountDownLatch(1) + val refreshes = AtomicInteger() + val gate = ProjectRequestGate { + if (refreshes.incrementAndGet() == 1) { + refreshing.countDown() + finishRefresh.awaitChecked() + } + } + val first = task { gate.run(true) { refreshes.get() } } + try { + refreshing.awaitChecked() + val second = task { gate.run(true) { refreshes.get() } } + second.awaitQueued() + finishRefresh.countDown() + assertEquals(1, first.result()) + assertEquals(2, second.result()) + } finally { + finishRefresh.countDown() + } + } + @Test fun `a reader arriving during active reads waits for its own refresh`() { val reading = CountDownLatch(1) @@ -75,31 +97,30 @@ class ProjectRequestGateTest { @Test fun `refresh failure is shared by its batch and releases admission`() { - val refreshing = CountDownLatch(1) - val finishRefresh = CountDownLatch(1) + val exclusiveRunning = CountDownLatch(1) + val finishExclusive = CountDownLatch(1) val refreshes = AtomicInteger() val failure = IllegalStateException("refresh failed") val gate = ProjectRequestGate { - if (refreshes.incrementAndGet() == 1) { - refreshing.countDown() - finishRefresh.awaitChecked() - throw failure - } + if (refreshes.incrementAndGet() == 2) throw failure } - val first = task { runCatching { gate.run(true) { fail("must not run") } }.exceptionOrNull() } + val running = task { gate.run(false) { exclusiveRunning.countDown(); finishExclusive.awaitChecked() } } try { - refreshing.awaitChecked() + exclusiveRunning.awaitChecked() + val first = task { runCatching { gate.run(true) { fail("must not run") } }.exceptionOrNull() } + first.awaitQueued() val second = task { runCatching { gate.run(true) { fail("must not run") } }.exceptionOrNull() } second.awaitQueued() val exclusive = task { gate.run(false) { "recovered" } } exclusive.awaitQueued() - finishRefresh.countDown() + finishExclusive.countDown() + running.result() assertSame(failure, first.result()) assertSame(failure, second.result()) assertEquals("recovered", exclusive.result()) - assertEquals(2, refreshes.get()) + assertEquals(3, refreshes.get()) } finally { - finishRefresh.countDown() + finishExclusive.countDown() } } diff --git a/daemon/src/test/kotlin/com/specificlanguages/mops/daemon/SharedMpsEnvironment.kt b/daemon/src/test/kotlin/com/specificlanguages/mops/daemon/SharedMpsEnvironment.kt index ee4f7aa2..9f1d630e 100644 --- a/daemon/src/test/kotlin/com/specificlanguages/mops/daemon/SharedMpsEnvironment.kt +++ b/daemon/src/test/kotlin/com/specificlanguages/mops/daemon/SharedMpsEnvironment.kt @@ -142,6 +142,11 @@ object SharedMpsEnvironment { private fun boot(): IdeaEnvironment { applyMpsSystemProperties() + FileWatcherExecutable.install( + requiredPathProperty("test.mpsHome"), + Files.createTempDirectory("mops-fsnotifier"), + DaemonLogger(), + ) val handoff = SynchronousQueue() val environmentThread = Thread({ diff --git a/docs/daemon-request-concurrency.md b/docs/daemon-request-concurrency.md index 9c6554c5..b3eb6ad1 100644 --- a/docs/daemon-request-concurrency.md +++ b/docs/daemon-request-concurrency.md @@ -4,9 +4,13 @@ The daemon accepts connections concurrently. Authenticated pings return daemon m admission, so a long build, test run, or Code Mode program does not block discovery pings. Domain requests enter a FIFO queue outside MPS model actions. With no active request, the first queued caller refreshes -external changes exclusively. If it is a parallel reader, all consecutive readers queued when that refresh finishes -share the refreshed state and run concurrently. The first exclusive request ends the batch. Requests arriving after -admission wait for another refresh after the active batch finishes. There is no batching delay. +external changes exclusively. If it is a parallel reader, the consecutive readers that were queued when that refresh +started share the refreshed state and run concurrently. The first exclusive request ends the batch, and so does the +first request that arrived during the refresh, so every request sees the external changes made before it arrived. +Requests arriving later wait for another refresh after the active batch finishes. There is no batching delay. + +The refresh relies on the native file watcher when it can; see `WatchedFolderRefresh` for when it walks the folders +instead. The daemon log names the way each refresh went and its cost. ```text refresh → [list, get-node, find] → refresh → edit → refresh → [list, find] diff --git a/docs/test-running-design.md b/docs/test-running-design.md index fdcb8e49..078cda4b 100644 --- a/docs/test-running-design.md +++ b/docs/test-running-design.md @@ -135,9 +135,9 @@ synchronously when it opens a project; the test-mode platform skips that refresh from disk the project folder, the repository folders, and the module folders of the selection that are not inside another of these folders, and MPS reloads what changed. A recursive refresh checks only files loaded into memory, and a file loaded later keeps its stored timestamp, length, and content (IJPL-219404), so the worker first loads every file -the VFS has stored under these folders. The daemon refreshes its own persistent VFS the same way before each request. -IntelliJ's own version and corruption checks rebuild outdated or broken storage. mops adds no check of its own: deleting -the cache is the documented way out and costs one cold run. +the VFS has stored under these folders. The daemon refreshes its own persistent VFS the same way before its first +request. IntelliJ's own version and corruption checks rebuild outdated or broken storage. mops adds no check of its own: +deleting the cache is the documented way out and costs one cold run. The load includes generated output folders: generation compares new output with the content the VFS holds and skips unchanged files, and it finds and deletes stale files through the VFS. Test-worker class loading reads `deps.cp` from diff --git a/skills/mops-daemon/SKILL.md b/skills/mops-daemon/SKILL.md index 7dd1d61e..82b9983c 100644 --- a/skills/mops-daemon/SKILL.md +++ b/skills/mops-daemon/SKILL.md @@ -82,6 +82,7 @@ reported workspace rather than assuming a hash. A workspace holds: workspace path, and startup time. - `logs/daemon.log`: the daemon's startup and runtime log. - `daemon/config` and `daemon/system`: the isolated IDEA directories passed to the daemon JVM. +- `fsnotifier/`: the native file watcher executable copied from the MPS distribution. Deleting a live daemon's record or cache does not stop it. Stale records are removed when the recorded process or socket is no longer reachable. Commands reach the daemon over a loopback socket with a per-daemon token. @@ -106,4 +107,9 @@ directory under its `test-runs//` and says so in the report's diagnostic another directory for a newly started daemon. The cache is safe to delete while no test runs; delete it if test results contradict the sources, and to free the disk space of MPS builds no longer in use. +Before each request the daemon picks up external file changes. It relies on the native file watcher of the IntelliJ +platform and walks the project and module folders when the watcher is unavailable or cannot be trusted; each refresh +writes a line to the daemon log that names how it went. `MOPS_DAEMON_JAVA_OPTIONS=-Didea.filewatcher.disabled=true` +turns the watcher off. + Stop existing daemons after updating mops. From 2f19ccd90fe368579b05d60f8350ae7441156ed8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sergej=20Ko=C5=A1=C4=8Dejev?= Date: Thu, 5 Nov 2026 22:21:10 +0100 Subject: [PATCH 2/3] fix: fall back to a full walk when the fence cannot be written --- .../specificlanguages/mops/daemon/FileWatcherFence.kt | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/FileWatcherFence.kt b/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/FileWatcherFence.kt index 6b69e0b6..85ea2304 100644 --- a/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/FileWatcherFence.kt +++ b/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/FileWatcherFence.kt @@ -31,7 +31,7 @@ import kotlin.io.path.deleteIfExists internal class FileWatcherFence private constructor(private val root: Path) { private val pending = ConcurrentHashMap() - /** Returns `true` once the watcher reported the fence, or `false` after [timeout]. */ + /** Returns `true` once the watcher reported the fence, or `false` after [timeout] or when it cannot be written. */ fun pass(timeout: Duration): Boolean { val marker = root.resolve(UUID.randomUUID().toString()) val reported = CountDownLatch(1) @@ -40,6 +40,8 @@ internal class FileWatcherFence private constructor(private val root: Path) { Files.createDirectory(marker) Files.createFile(marker.resolve("fence")) return reported.await(timeout.toMillis(), TimeUnit.MILLISECONDS) + } catch (_: IOException) { + return false } finally { pending.remove(marker.toString()) deleteQuietly(marker) @@ -74,7 +76,11 @@ internal class FileWatcherFence private constructor(private val root: Path) { } if (hook.get(fileWatcher) != null) return null // Reported paths are canonical and system-dependent, so the root is too. - val root = Path.of(PathManager.getSystemPath(), "mops-file-watcher-fence").createDirectories().toRealPath() + val root = try { + Path.of(PathManager.getSystemPath(), "mops-file-watcher-fence").createDirectories().toRealPath() + } catch (_: IOException) { + return null + } root.toFile().listFiles()?.forEach { FileUtil.delete(it) } LocalFileSystem.getInstance().addRootToWatch(FileUtil.toSystemIndependentName(root.toString()), true) val fence = FileWatcherFence(root) From 83dd27754fda01eb3853f6add2f5bdb6330c7935 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sergej=20Ko=C5=A1=C4=8Dejev?= Date: Thu, 5 Nov 2026 22:30:49 +0100 Subject: [PATCH 3/3] refactor: address review findings for the watched folder refresh --- .../mops/daemon/FileWatcherFence.kt | 14 ++++---- .../mops/daemon/ProjectRequestGate.kt | 13 ++++---- .../mops/daemon/WatchedFolderRefresh.kt | 32 +++++++++++++------ .../daemon/FileWatcherRefreshSemanticsTest.kt | 9 ++++-- 4 files changed, 42 insertions(+), 26 deletions(-) diff --git a/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/FileWatcherFence.kt b/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/FileWatcherFence.kt index 85ea2304..80e549e0 100644 --- a/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/FileWatcherFence.kt +++ b/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/FileWatcherFence.kt @@ -19,20 +19,20 @@ import kotlin.io.path.createDirectories import kotlin.io.path.deleteIfExists /** - * Makes sure the native file watcher has reported every change made before [pass] was called. There is no API to + * Makes sure the native file watcher has reported every change made before [awaitReported] was called. There is no API to * flush pending watcher events, and on macOS an event arrives up to about 300 ms after the change. `fsnotifier` reports - * events in order, so once it reports a change made by [pass], it has reported every earlier change too. + * events in order, so once it reports a change made by [awaitReported], it has reported every earlier change too. * - * Each pass creates a fresh directory with a file in it under [root], a folder that is watched recursively, and waits + * Each call creates a fresh directory with a file in it under [root], a folder that is watched recursively, and waits * until the watcher reports a path inside that directory. A fresh directory can only be named by events that happened * after it was created. The raw watcher events are observed through `FileWatcher.myTestNotifier`, a private hook that * the platform calls for every event; see the mps-api-research note native-file-watcher-in-headless-mps.md. */ -internal class FileWatcherFence private constructor(private val root: Path) { +internal class FileWatcherFence private constructor(val root: Path) { private val pending = ConcurrentHashMap() /** Returns `true` once the watcher reported the fence, or `false` after [timeout] or when it cannot be written. */ - fun pass(timeout: Duration): Boolean { + fun awaitReported(timeout: Duration): Boolean { val marker = root.resolve(UUID.randomUUID().toString()) val reported = CountDownLatch(1) pending[marker.toString()] = reported @@ -59,7 +59,7 @@ internal class FileWatcherFence private constructor(private val root: Path) { marker.resolve("fence").deleteIfExists() marker.deleteIfExists() } catch (_: IOException) { - // A leftover fence directory is harmless: every pass uses a new one. + // A leftover fence directory is harmless: every call uses a new one. } } @@ -68,7 +68,7 @@ internal class FileWatcherFence private constructor(private val root: Path) { val instance: FileWatcherFence? by lazy { create() } private fun create(): FileWatcherFence? { - val fileWatcher = (LocalFileSystem.getInstance() as? LocalFileSystemImpl)?.fileWatcher ?: return null + val fileWatcher = (LocalFileSystem.getInstance() as LocalFileSystemImpl).fileWatcher val hook = try { FileWatcher::class.java.getDeclaredField("myTestNotifier").apply { isAccessible = true } } catch (_: ReflectiveOperationException) { diff --git a/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/ProjectRequestGate.kt b/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/ProjectRequestGate.kt index c116bdfc..0eb658a7 100644 --- a/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/ProjectRequestGate.kt +++ b/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/ProjectRequestGate.kt @@ -14,16 +14,16 @@ internal class ProjectRequestGate(private val refresh: () -> Unit) { private val waiting = ArrayDeque() private var active = 0 private var refreshing = false - private var queued = 0L + private var issuedTickets = 0L fun run(parallelRead: Boolean, operation: () -> T): T { val ticket = lock.withLock { - val ticket = Ticket(parallelRead, ++queued) + val ticket = Ticket(parallelRead, ++issuedTickets) waiting.addLast(ticket) while (!ticket.admitted) { if (active == 0 && !refreshing && waiting.first() === ticket) { refreshing = true - val queuedBeforeRefresh = queued + val issuedBeforeRefresh = issuedTickets lock.unlock() val failure = try { refresh() @@ -40,7 +40,7 @@ internal class ProjectRequestGate(private val refresh: () -> Unit) { next.failure = failure next.admitted = true active++ - } while (parallelRead && waiting.firstOrNull()?.sharesRefresh(queuedBeforeRefresh) == true) + } while (parallelRead && waiting.firstOrNull()?.sharesRefresh(issuedBeforeRefresh) == true) refreshing = false changed.signalAll() } else { @@ -60,8 +60,9 @@ internal class ProjectRequestGate(private val refresh: () -> Unit) { } } - private class Ticket(val parallelRead: Boolean, val number: Long) { - fun sharesRefresh(queuedBeforeRefresh: Long) = parallelRead && number <= queuedBeforeRefresh + private class Ticket(val parallelRead: Boolean, val sequence: Long) { + /** Whether this ticket joins a batch of readers whose refresh started after [issuedBeforeRefresh] tickets. */ + fun sharesRefresh(issuedBeforeRefresh: Long) = parallelRead && sequence <= issuedBeforeRefresh var admitted = false var failure: Throwable? = null diff --git a/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/WatchedFolderRefresh.kt b/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/WatchedFolderRefresh.kt index 741aea3d..84c4a434 100644 --- a/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/WatchedFolderRefresh.kt +++ b/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/WatchedFolderRefresh.kt @@ -17,8 +17,9 @@ import java.time.Duration * makes sure the watcher has reported every change made before the request, and a non-forced recursive refresh then * visits only the files the watcher reported. The full walk stays the fallback: when the watcher is not operational or * has reported a failure, when the fence times out, and on the first refresh after the watch roots changed or - * `fsnotifier` was started again. A folder is also walked fully the first time it is refreshed and whenever it lies - * outside every recursive watch root. + * `fsnotifier` was started again. A folder is also walked fully the first time it is refreshed, whenever it lies + * outside every recursive watch root, and on Windows when it is on another drive than the fence: `fsnotifier` watches + * each drive in its own thread, so it keeps the order of events only within a drive. * * See the mps-api-research note native-file-watcher-in-headless-mps.md. */ @@ -31,17 +32,21 @@ internal class WatchedFolderRefresh { val started = System.nanoTime() val watcher = HeadlessWatcher.find() val fileWatcher = (LocalFileSystem.getInstance() as LocalFileSystemImpl).fileWatcher + // Created before the roots settle, because creating it adds a watch root. + val fence = if (watcher != null && fileWatcher.isOperational) FileWatcherFence.instance else null val reason = when { - watcher == null || !fileWatcher.isOperational -> "the file watcher is not operational" + watcher == null -> "the mops file watcher is not registered" + !fileWatcher.isOperational -> "the file watcher is not operational" !awaitSettledRoots(fileWatcher::isSettingRoots) -> "the file watcher is still setting its roots" - else -> fenceAndCompare(watcher) + else -> fenceAndCompare(watcher, fence) } - return Pass(reason, watcher?.recursiveRoots().orEmpty(), Duration.ofNanos(System.nanoTime() - started)) + return Pass( + reason, watcher?.recursiveRoots().orEmpty(), fence?.root, Duration.ofNanos(System.nanoTime() - started), + ) } - private fun fenceAndCompare(watcher: HeadlessWatcher): String? { - val fence = FileWatcherFence.instance - val fenced = fence?.pass(FENCE_TIMEOUT) ?: false + private fun fenceAndCompare(watcher: HeadlessWatcher, fence: FileWatcherFence?): String? { + val fenced = fence?.awaitReported(FENCE_TIMEOUT) ?: false // The watcher reports failures as notifications that are delivered on the EDT. ApplicationManager.getApplication().invokeAndWait {} val changeCount = watcher.changeCount() @@ -70,7 +75,12 @@ internal class WatchedFolderRefresh { * The refresh of one request. [fullWalkReason] is `null` when the watcher is trusted and says otherwise why every * folder is walked. */ - inner class Pass(val fullWalkReason: String?, private val recursiveRoots: List, private val preparation: Duration) { + inner class Pass( + val fullWalkReason: String?, + private val recursiveRoots: List, + private val fenceRoot: Path?, + private val preparation: Duration, + ) { private var walked = 0 private var watched = 0 private var refreshing = Duration.ZERO @@ -96,7 +106,8 @@ internal class WatchedFolderRefresh { } catch (_: IOException) { return false } - return recursiveRoots.any(real::startsWith) + val sameDrive = !isWindows || real.root == fenceRoot?.root + return sameDrive && recursiveRoots.any(real::startsWith) } /** A line for the daemon log that names how the folders were refreshed and what it cost. */ @@ -127,5 +138,6 @@ internal class WatchedFolderRefresh { val FENCE_TIMEOUT: Duration = Duration.ofSeconds(2) val ROOTS_TIMEOUT: Duration = Duration.ofSeconds(10) const val POLL_MILLIS = 10L + val isWindows = System.getProperty("os.name").startsWith("Windows") } } diff --git a/daemon/src/test/kotlin/com/specificlanguages/mops/daemon/FileWatcherRefreshSemanticsTest.kt b/daemon/src/test/kotlin/com/specificlanguages/mops/daemon/FileWatcherRefreshSemanticsTest.kt index 292d49b2..2f54a59a 100644 --- a/daemon/src/test/kotlin/com/specificlanguages/mops/daemon/FileWatcherRefreshSemanticsTest.kt +++ b/daemon/src/test/kotlin/com/specificlanguages/mops/daemon/FileWatcherRefreshSemanticsTest.kt @@ -15,7 +15,7 @@ import kotlin.test.assertNull /** * A refresh sees an external edit made immediately before it, also when it relies on the native file watcher instead - * of walking the folders. On macOS the watcher must be in use, so these tests exercise the fence. + * of walking the folders. On macOS and on Linux x86-64 the watcher must be in use, so these tests exercise the fence. */ class FileWatcherRefreshSemanticsTest { @@ -59,14 +59,17 @@ class FileWatcherRefreshSemanticsTest { val report = refresh.refresh() - if (isMacOs) assertNull(report.fullWalkReason, "edit $index: the refresh relies on the file watcher") + if (watcherExpected) assertNull(report.fullWalkReason, "edit $index: the refresh relies on the file watcher") assertEquals(name, propertyValue(access.read(node), "name"), "edit $index is visible after the refresh") } } private companion object { const val EDITS = 20 - val isMacOs = System.getProperty("os.name").startsWith("Mac") + // Every supported MPS version ships fsnotifier for macOS and for Linux on x86-64. + val watcherExpected = System.getProperty("os.name").let { os -> + os.startsWith("Mac") || os.startsWith("Linux") && System.getProperty("os.arch") in setOf("amd64", "x86_64") + } const val STRUCTURE_MODEL_NAME = "com.specificlanguages.json.structure" const val STRUCTURE_MODEL_PATH =