-
Notifications
You must be signed in to change notification settings - Fork 46
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
There is often noticeable lag between when a file event happens and when the build is triggered. There is also often latency when the user presses enter to interrupt the watch. Both are due to the way that SourceModificationWatch polls for results. It can be removed by having the main thread block on an event queue that is asynchronously filled by threads that wait for user input and file events. I made a similar change in CloseWatch. The new EventMonitor is where most of the logic lives. It is implemented as a block box that effectively provides three methods: 1) watch() -- block the main thread until a user input or watched file event occurs, returning true if the build was triggered 2) state() -- a snapshot of the current watch state. This is primarily for legacy compatibility with sbt, which has a number of methods that take the state as input. In practice, all that sbt really needs most of the time is the count of the number of build triggers 3) close() -- shutdown any threads or service started up by the event monitor I implemented the EventMonitor as a block box so that it would be straightforward to change the existing implementation or add new implementations without having to break forward and backwards compatibility with sbt. In particular, I can envision adding a second EventMonitor that uses a persistent file system cache for file events instead of the ad-hoc cache that currently exists in the EventMonitorImpl.eventThread. At the moment, the one implementation is EventMonitorImpl. It spins up two threads for monitoring user input and the file events. Both write `EventMonitor.Events` to a concurrent queue that the main thread reads from. User input events supersede file triggers. When the user hits enter, the queue is cleared and filled with the exit event. Otherwise, the event thread polls the watch service for events. When it receives a file event, it adds the event to a cache of recent events. This cache is used to prevent multiple builds from being triggered by the same file*. The eventThread also detects when directories are created or deleted and registers or unregisters the directory with the watch service as needed. I also stopped registering all of the files. Only directories are registered now. The registered files just waste memory. This commit also adds logging of watch events. Unfortunately, I can't get the logging to actually work at the debug level due to sbt/sbt#4097, but once that issue is fixed, logging with the new EventMonitor should work. I added a simple test that the anti-entropy works and made some small adjustments to the existing tests to make them work with the new implementation. For backwards compatibility with older versions of sbt, I re-implement SourceModificationWatch.watch to wrap an EventMonitor that we shutdown after each trigger. This is also the reason that I had to add two close methods to EventMonitor -- one that shuts down the watch service, and one that doesn't. * When neovim, for example, saves a file. It moves the buffer into the file location which can trigger a delete and a create event on the file. Without the anti-entropy timeout, there would be two build triggered for the same file save.
- Loading branch information
Showing
5 changed files
with
362 additions
and
103 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,194 @@ | ||
| package sbt.internal.io | ||
|
|
||
| import java.nio.file.{ ClosedWatchServiceException, Files, Path, WatchKey } | ||
| import java.util.concurrent.ArrayBlockingQueue | ||
| import java.util.concurrent.atomic.{ AtomicBoolean, AtomicInteger } | ||
|
|
||
| import sbt.io.WatchService | ||
|
|
||
| import scala.annotation.tailrec | ||
| import scala.collection.JavaConverters._ | ||
| import scala.concurrent.duration._ | ||
|
|
||
| private[sbt] sealed trait EventMonitor extends AutoCloseable { | ||
|
|
||
| /** A snapshot of the WatchState that includes the number of build triggers and watch sources. */ | ||
| def state(): WatchState | ||
|
|
||
| /** Block indefinitely until the monitor receives a file event or the user stops the watch. */ | ||
| def watch(): Boolean | ||
|
|
||
| /** Cleans up any service and/or threads started by the monitor */ | ||
| override def close(): Unit = close(closeService = true) | ||
| /* | ||
| * Workaround for the legacy implementation of SourceModificationWatch.watch | ||
| */ | ||
| private[io] def close(closeService: Boolean): Unit | ||
| } | ||
|
|
||
| object EventMonitor { | ||
| private sealed trait Event | ||
| private case object Cancelled extends Event | ||
| private case class Triggered(path: Path) extends Event | ||
|
|
||
| private class EventMonitorImpl private[EventMonitor] ( | ||
| private[this] val service: WatchService, | ||
| private[this] val events: ArrayBlockingQueue[Event], | ||
| private[this] val eventThread: Looper with HasWatchState, | ||
| private[this] val userInputThread: Looper, | ||
| private[this] val logger: Logger) | ||
| extends EventMonitor { | ||
|
|
||
| override def state(): WatchState = eventThread.state() | ||
|
|
||
| override def watch(): Boolean = events.take() match { | ||
| case Cancelled => false | ||
| case Triggered(path) => | ||
| logger.debug(s"Triggered watch event due to updated path: $path") | ||
| eventThread.incrementCount() | ||
| true | ||
| } | ||
|
|
||
| override def close(closeState: Boolean): Unit = { | ||
| if (closed.compareAndSet(false, true)) { | ||
| if (closeState) service.close() | ||
| userInputThread.close() | ||
| eventThread.close() | ||
| logger.debug("Closed EventMonitor") | ||
| } | ||
| } | ||
|
|
||
| private[this] val closed = new AtomicBoolean(false) | ||
| } | ||
|
|
||
| def apply(state: WatchState, | ||
| delay: FiniteDuration, | ||
| antiEntropy: FiniteDuration, | ||
| terminationCondition: => Boolean, | ||
| logger: Logger = NullLogger): EventMonitor = { | ||
| val events = new ArrayBlockingQueue[Event](1) | ||
| val eventThread = newEventsThread(delay, antiEntropy, state, events, logger) | ||
| val userInputThread = newUserInputThread(terminationCondition, events, logger) | ||
| new EventMonitorImpl(state.service, events, eventThread, userInputThread, logger) | ||
| } | ||
|
|
||
| private trait HasWatchState { | ||
| def state(): WatchState | ||
| def incrementCount(): Unit | ||
| } | ||
| private def newEventsThread(delay: FiniteDuration, | ||
| antiEntropy: FiniteDuration, | ||
| s: WatchState, | ||
| events: ArrayBlockingQueue[Event], | ||
| logger: Logger): Looper with HasWatchState = { | ||
| var recentEvents = Map.empty[Path, Deadline] | ||
| new Looper(s"watch-state-event-thread-${eventThreadId.incrementAndGet()}") with HasWatchState { | ||
| private[this] val lock = new Object | ||
| private[this] var count = s.count | ||
| private[this] var registered = s.registered | ||
| def incrementCount(): Unit = lock.synchronized { count += 1 } | ||
| def state(): WatchState = lock.synchronized(s.withCount(count).withRegistered(registered)) | ||
| override def loop(): Unit = { | ||
| recentEvents = recentEvents.filterNot(_._2.isOverdue) | ||
| getFilesForKey(s.service.poll(delay)).foreach(maybeTrigger) | ||
| } | ||
| def getFilesForKey(key: WatchKey): Seq[Path] = key match { | ||
| case null => Nil | ||
| case k => | ||
| val allEvents = k.pollEvents.asScala | ||
| .map(e => k.watchable.asInstanceOf[Path].resolve(e.context.asInstanceOf[Path])) | ||
| logger.debug(s"Received events:\n${allEvents.mkString("\n")}") | ||
| val (exist, notExist) = allEvents.partition(Files.exists(_)) | ||
| val (updatedDirectories, updatedFiles) = exist.partition(Files.isDirectory(_)) | ||
| val newFiles = updatedDirectories.flatMap(filesForNewDirectory) | ||
| lock.synchronized { registered --= notExist } | ||
| notExist.foreach(s.unregister) | ||
| updatedFiles ++ newFiles ++ notExist | ||
| } | ||
| /* | ||
| * Returns new files found in new directory and any subdirectories, assuming that there is | ||
| * a recursive source with a base that is parent to the directory. | ||
| */ | ||
| def filesForNewDirectory(dir: Path): Seq[Path] = { | ||
| lazy val recursive = | ||
| s.sources.exists(src => dir.startsWith(src.base.toPath) && src.recursive) | ||
| if (!registered.contains(dir) && recursive) { | ||
| val dirs = Files.walk(dir).iterator.asScala.filter(Files.isDirectory(_)) | ||
| val newDirs = dirs.map(d => d -> s.register(d)).toIndexedSeq | ||
| lock.synchronized { registered ++= newDirs } | ||
| Files.walk(dir).iterator.asScala.toSeq | ||
| } else Nil | ||
| } | ||
| /* | ||
| * Triggers only if there is no pending Trigger and the file is not in an anti-entropy | ||
| * quarantine. | ||
| */ | ||
| def maybeTrigger(path: Path): Unit = | ||
| if (s.accept(path)) { | ||
| if (recentEvents.get(path).fold(false)(!_.isOverdue)) | ||
| logger.debug(s"Ignoring watch event for $path due to anti-entropy constraint") | ||
| else | ||
| events.peek() match { | ||
| case Cancelled => | ||
| logger.debug(s"Watch cancelled, not offering event for path $path") | ||
| case _ => | ||
| recentEvents += path -> antiEntropy.fromNow | ||
| if (!events.offer(Triggered(path))) { | ||
| logger.debug(s"Event already pending, dropping event for path: $path") | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
| // Shutup the compiler about unused arguments | ||
| @inline private[this] def ignoreArg(arg: => Any): Unit = if (true) () else { arg; () } | ||
| trait Logger { | ||
| def debug(msg: => Any): Unit = ignoreArg(msg) | ||
| } | ||
| object NullLogger extends Logger | ||
| private def newUserInputThread(terminationCondition: => Boolean, | ||
| events: ArrayBlockingQueue[Event], | ||
| logger: Logger): Looper = | ||
| new Looper(s"watch-state-user-input-${userInputId.incrementAndGet}") { | ||
| override final def loop(): Unit = { | ||
| if (terminationCondition) { | ||
| logger.debug("Received termination condition. Stopping watch...") | ||
| while (!events.offer(Cancelled)) { | ||
| events.clear() | ||
| } | ||
| } else {} | ||
| } | ||
| } | ||
|
|
||
| private abstract class Looper(name: String) extends Thread(name) with AutoCloseable { | ||
| private[this] var stopped = false | ||
| def isStopped: Boolean = this.synchronized(stopped) | ||
| def loop(): Unit | ||
| @tailrec | ||
| override final def run(): Unit = { | ||
| try { | ||
| if (!isStopped) { | ||
| loop() | ||
| } | ||
| } catch { | ||
| case (_: ClosedWatchServiceException | _: InterruptedException) => | ||
| this.synchronized { stopped = true } | ||
| } | ||
| if (!isStopped) { | ||
| run() | ||
| } | ||
| } | ||
| def close(): Unit = this.synchronized { | ||
| if (!stopped) { | ||
| stopped = true | ||
| this.interrupt() | ||
| this.join(5000) | ||
| } | ||
| } | ||
| setDaemon(true) | ||
| start() | ||
| } | ||
| private val eventThreadId = new AtomicInteger(0) | ||
| private val userInputId = new AtomicInteger(0) | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.