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..80e549e0 --- /dev/null +++ b/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/FileWatcherFence.kt @@ -0,0 +1,91 @@ +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 [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 [awaitReported], it has reported every earlier change too. + * + * 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(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 awaitReported(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) + } catch (_: IOException) { + return false + } 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 call 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 + 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 = 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) + 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 c5cbe098..9ed8e028 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 728b90c6..7520fba8 100644 --- a/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/MopsDaemonCommand.kt +++ b/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/MopsDaemonCommand.kt @@ -67,6 +67,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..0eb658a7 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 issuedTickets = 0L fun run(parallelRead: Boolean, operation: () -> T): T { - val ticket = Ticket(parallelRead) - lock.withLock { + val ticket = lock.withLock { + val ticket = Ticket(parallelRead, ++issuedTickets) waiting.addLast(ticket) while (!ticket.admitted) { if (active == 0 && !refreshing && waiting.first() === ticket) { refreshing = true + val issuedBeforeRefresh = issuedTickets 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(issuedBeforeRefresh) == true) refreshing = false changed.signalAll() } else { changed.awaitUninterruptibly() } } + ticket } try { ticket.failure?.let { throw it } @@ -52,7 +60,10 @@ internal class ProjectRequestGate(private val refresh: () -> Unit) { } } - private class Ticket(val parallelRead: Boolean) { + 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 new file mode 100644 index 00000000..84c4a434 --- /dev/null +++ b/daemon/src/main/kotlin/com/specificlanguages/mops/daemon/WatchedFolderRefresh.kt @@ -0,0 +1,143 @@ +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, 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. + */ +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 + // 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 -> "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, fence) + } + return Pass( + reason, watcher?.recursiveRoots().orEmpty(), fence?.root, Duration.ofNanos(System.nanoTime() - started), + ) + } + + 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() + 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 fenceRoot: Path?, + 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 + } + 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. */ + 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 + 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 new file mode 100644 index 00000000..2f54a59a --- /dev/null +++ b/daemon/src/test/kotlin/com/specificlanguages/mops/daemon/FileWatcherRefreshSemanticsTest.kt @@ -0,0 +1,83 @@ +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 and on Linux x86-64 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 (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 + // 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 = + "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 84212d6a..8f0afb37 100644 --- a/daemon/src/test/kotlin/com/specificlanguages/mops/daemon/SharedMpsEnvironment.kt +++ b/daemon/src/test/kotlin/com/specificlanguages/mops/daemon/SharedMpsEnvironment.kt @@ -141,6 +141,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/skills/mops-daemon/SKILL.md b/skills/mops-daemon/SKILL.md index 9ed211fd..08fcece2 100644 --- a/skills/mops-daemon/SKILL.md +++ b/skills/mops-daemon/SKILL.md @@ -85,6 +85,7 @@ holds: home, 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. - `test-worker-system`: the IntelliJ system directory of the test worker. Only one daemon runs per project. When a command needs a daemon for another MPS home or build than the running one, mops @@ -112,4 +113,9 @@ Test runs keep the IntelliJ system directory (VFS and indexes) of the test worke (timed out, cancelled, or a crashed worker) deletes it, and the next run indexes again. The directory is safe to delete while no test runs; delete it if test results contradict the sources. +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.