WIP: fs-watching

Author: arturaz

build.mill

@@ -159,7 +159,9 @@ object Deps {
   val junitInterface = mvn"com.github.sbt:junit-interface:0.13.3"
   val commonsIo = mvn"commons-io:commons-io:2.18.0"
   val log4j2Core = mvn"org.apache.logging.log4j:log4j-core:2.24.3"
-  val osLib = mvn"com.lihaoyi::os-lib:0.11.5-M2"
+  val osLibVersion = "0.11.5-M2"
+  val osLib = mvn"com.lihaoyi::os-lib:${osLibVersion}"
+  val osLibWatch = mvn"com.lihaoyi::os-lib-watch:${osLibVersion}"
   val pprint = mvn"com.lihaoyi::pprint:0.9.0"
   val mainargs = mvn"com.lihaoyi::mainargs:0.7.6"
   val millModuledefsVersion = "0.11.3-M5"

core/api/src/mill/api/Watchable.scala

@@ -8,6 +8,23 @@ package mill.api
  */
 private[mill] sealed trait Watchable
 private[mill] object Watchable {
-  case class Path(p: java.nio.file.Path, quick: Boolean, signature: Int) extends Watchable
-  case class Value(f: () => Long, signature: Long, pretty: String) extends Watchable
+  /** A [[Watchable]] that is being watched via polling. */
+  private[mill] sealed trait Pollable extends Watchable
+
+  /** A [[Watchable]] that is being watched via a notification system (like inotify). */
+  private[mill] sealed trait Notifiable extends Watchable
+
+  /**
+   * @param p the path to watch
+   * @param quick if true, only watch file attributes
+   * @param signature the initial hash of the path contents
+   */
+  case class Path(p: java.nio.file.Path, quick: Boolean, signature: Int) extends Notifiable
+
+  /**
+   * @param f the expression to watch, returns some sort of hash
+   * @param signature the initial hash from the first invocation of the expression
+   * @param pretty human-readable name
+   */
+  case class Value(f: () => Long, signature: Long, pretty: String) extends Pollable
 }

libs/main/src/mill/main/MainModule.scala

@@ -22,8 +22,8 @@ abstract class MainRootModule()(implicit
  * [[show]], [[inspect]], [[plan]], etc.
  */
 trait MainModule extends BaseModule with MainModuleApi {
-  protected[mill] val watchedValues: mutable.Buffer[Watchable] = mutable.Buffer.empty[Watchable]
-  protected[mill] val evalWatchedValues: mutable.Buffer[Watchable] = mutable.Buffer.empty[Watchable]
+  protected[mill] val watchedValues: mutable.Buffer[Watchable] = mutable.Buffer.empty
+  protected[mill] val evalWatchedValues: mutable.Buffer[Watchable] = mutable.Buffer.empty
   object interp {
     def watchValue[T](v0: => T)(implicit fn: sourcecode.FileName, ln: sourcecode.Line): T = {
       val v = v0

runner/package.mill

@@ -18,6 +18,7 @@ object `package` extends build.MillPublishScalaModule {
   def mvnDeps = Seq(
     build.Deps.sourcecode,
     build.Deps.osLib,
+    build.Deps.osLibWatch,
     build.Deps.mainargs,
     build.Deps.upickle,
     build.Deps.pprint,

runner/src/mill/runner/MillBuildBootstrap.scala

@@ -277,7 +277,8 @@ class MillBuildBootstrap(
         // look at the `moduleWatched` of one frame up (`prevOuterFrameOpt`),
         // and not the `moduleWatched` from the current frame (`prevFrameOpt`)
         val moduleWatchChanged =
-          prevOuterFrameOpt.exists(_.moduleWatched.exists(w => !Watching.validate(w)))
+          // QUESTION: is polling appropriate here?
+          prevOuterFrameOpt.exists(_.moduleWatched.exists(w => !Watching.validateAnyWatchable(w)))
 
         val classLoader = if (runClasspathChanged || moduleWatchChanged) {
           // Make sure we close the old classloader every time we create a new

runner/src/mill/runner/MillMain.scala

@@ -352,9 +352,8 @@ object MillMain {
                       if (config.watch.value) os.remove(out / OutFiles.millSelectiveExecution)
                       Watching.watchLoop(
                         ringBell = config.ringBell.value,
-                        watch = config.watch.value,
+                        watch = Option.when(config.watch.value)(Watching.WatchArgs(setIdle = setIdle, colors)),
                         streams = streams,
-                        setIdle = setIdle,
                         evaluate = (enterKeyPressed: Boolean, prevState: Option[RunnerState]) => {
                           adjustJvmProperties(userSpecifiedProperties, initialSystemProperties)
                           runMillBootstrap(
@@ -363,8 +362,7 @@ object MillMain {
                             config.leftoverArgs.value,
                             streams
                           )
-                        },
-                        colors = colors
+                        }
                       )
                     }
                   }

runner/src/mill/runner/Watching.scala

@@ -1,13 +1,14 @@
 package mill.runner
 
-import mill.internal.Colors
-import mill.define.internal.Watchable
-import mill.define.PathRef
 import mill.api.SystemStreams
 import mill.api.internal.internal
+import mill.define.PathRef
+import mill.define.internal.Watchable
+import mill.internal.Colors
 
 import java.io.InputStream
 import scala.annotation.tailrec
+import scala.util.Using
 
 /**
  * Logic around the "watch and wait" functionality in Mill: re-run on change,
@@ -17,40 +18,68 @@ import scala.annotation.tailrec
 object Watching {
   case class Result[T](watched: Seq[Watchable], error: Option[String], result: T)
 
+  trait Evaluate[T] {
+    def apply(enterKeyPressed: Boolean, previousState: Option[T]): Result[T]
+  }
+
+  case class WatchArgs(
+      setIdle: Boolean => Unit,
+      colors: Colors
+  )
+
+  /**
+   * @param ringBell whether to emit bells
+   * @param watch if false just runs once and returns
+   */
   def watchLoop[T](
       ringBell: Boolean,
-      watch: Boolean,
+      watch: Option[WatchArgs],
       streams: SystemStreams,
-      setIdle: Boolean => Unit,
-      evaluate: (Boolean, Option[T]) => Result[T],
-      colors: Colors
+      evaluate: Evaluate[T]
   ): (Boolean, T) = {
-    var prevState: Option[T] = None
-    var enterKeyPressed = false
-    while (true) {
-      val Result(watchables, errorOpt, result) = evaluate(enterKeyPressed, prevState)
-      prevState = Some(result)
+    def handleError(errorOpt: Option[String]): Unit = {
       errorOpt.foreach(streams.err.println)
-      if (ringBell) {
-        if (errorOpt.isEmpty) println("\u0007")
-        else {
-          println("\u0007")
-          Thread.sleep(250)
-          println("\u0007")
-        }
-      }
+      doRingBell(hasError = errorOpt.isDefined)
+    }
 
-      if (!watch) {
-        return (errorOpt.isEmpty, result)
-      }
+    def doRingBell(hasError: Boolean): Unit = {
+      if (!ringBell) return
 
-      val alreadyStale = watchables.exists(w => !validate(w))
-      enterKeyPressed = false
-      if (!alreadyStale) {
-        enterKeyPressed = Watching.watchAndWait(streams, setIdle, streams.in, watchables, colors)
+      println("\u0007")
+      if (hasError) {
+        // If we have an error ring the bell again
+        Thread.sleep(250)
+        println("\u0007")
       }
     }
-    ???
+
+    watch match {
+      case None =>
+        val Result(watchables, errorOpt, result) =
+          evaluate(enterKeyPressed = false, previousState = None)
+        handleError(errorOpt)
+        (errorOpt.isEmpty, result)
+
+      case Some(watchArgs) =>
+        var prevState: Option[T] = None
+        var enterKeyPressed = false
+
+        while (true) {
+          val Result(watchables, errorOpt, result) = evaluate(enterKeyPressed, prevState)
+          prevState = Some(result)
+          handleError(errorOpt)
+
+          // Do not enter watch if already stale, re-evaluate instantly.
+          val alreadyStale = watchables.exists(w => !validateAnyWatchable(w))
+          if (alreadyStale) {
+            enterKeyPressed = false
+          } else {
+            enterKeyPressed = watchAndWait(streams, watchArgs.setIdle, streams.in, watchables, watchArgs.colors)
+          }
+        }
+        // QUESTION: this never exits?
+        throw new IllegalStateException("unreachable")
+    }
   }
 
   def watchAndWait(
@@ -61,28 +90,50 @@ object Watching {
       colors: Colors
   ): Boolean = {
     setIdle(true)
-    val watchedPaths = watched.collect { case p: Watchable.Path => p.p }
-    val watchedValues = watched.size - watchedPaths.size
+    val (watchedPollables, watchedPaths) = watched.partitionMap {
+      case w: Watchable.Pollable => Left(w)
+      case p: Watchable.Path => Right(p)
+    }
+    val watchedValueCount = watched.size - watchedPaths.size
 
-    val watchedValueStr = if (watchedValues == 0) "" else s" and $watchedValues other values"
+    val watchedValueStr =
+      if (watchedValueCount == 0) "" else s" and $watchedValueCount other values"
 
     streams.err.println(
       colors.info(
         s"Watching for changes to ${watchedPaths.size} paths$watchedValueStr... (Enter to re-run, Ctrl-C to exit)"
       ).toString
     )
 
-    val enterKeyPressed = statWatchWait(watched, stdin)
-    setIdle(false)
-    enterKeyPressed
+    @volatile var pathChangesDetected = false
+    Using.resource(os.watch.watch(
+      watchedPaths.map(path => os.Path(path.p)),
+      onEvent = _ => pathChangesDetected = true,
+      logger = (eventType, data) => {
+        streams.out.println(colors.info(s"[watch] $eventType: ${pprint.apply(data)}"))
+      }
+    )) { _ =>
+      val enterKeyPressed =
+        statWatchWait(watchedPollables, stdin, notifiablesChanged = () => pathChangesDetected)
+      setIdle(false)
+      enterKeyPressed
+    }
   }
 
-  // Returns `true` if enter key is pressed to re-run tasks explicitly
-  def statWatchWait(watched: Seq[Watchable], stdin: InputStream): Boolean = {
+  /**
+   * @param notifiablesChanged returns true if any of the notifiables have changed
+   *
+   * @return `true` if enter key is pressed to re-run tasks explicitly, false if changes in watched files occured.
+   */
+  def statWatchWait(
+      watched: Seq[Watchable.Pollable],
+      stdin: InputStream,
+      notifiablesChanged: () => Boolean
+  ): Boolean = {
     val buffer = new Array[Byte](4 * 1024)
 
     @tailrec def statWatchWait0(): Boolean = {
-      if (watched.forall(w => validate(w))) {
+      if (!notifiablesChanged() && watched.forall(w => validate(w))) {
         if (lookForEnterKey()) {
           true
         } else {
@@ -96,11 +147,13 @@ object Watching {
       if (stdin.available() == 0) false
       else stdin.read(buffer) match {
         case 0 | -1 => false
-        case n =>
+        case bytesRead =>
           buffer.indexOf('\n') match {
             case -1 => lookForEnterKey()
-            case i =>
-              if (i >= n) lookForEnterKey()
+            case index =>
+              // If we found the newline further than the bytes read, that means it's not from this read and thus we
+              // should try reading again.
+              if (index >= bytesRead) lookForEnterKey()
               else true
           }
       }
@@ -109,13 +162,23 @@ object Watching {
     statWatchWait0()
   }
 
-  def validate(w: Watchable) = poll(w) == signature(w)
-  def poll(w: Watchable) = w match {
+  /** @return true if the watchable did not change. */
+  inline def validate(w: Watchable.Pollable): Boolean = validateAnyWatchable(w)
+
+  /**
+   * As [[validate]] but accepts any [[Watchable]] for the cases when we do not want to use a notification system.
+   *
+   * Normally you should use [[validate]] so that types would guide your implementation.
+   */
+  def validateAnyWatchable(w: Watchable): Boolean = poll(w) == signature(w)
+
+  def poll(w: Watchable): Long = w match {
     case Watchable.Path(p, quick, sig) =>
       new PathRef(os.Path(p), quick, sig, PathRef.Revalidate.Once).recomputeSig()
     case Watchable.Value(f, sig, pretty) => f()
   }
-  def signature(w: Watchable) = w match {
+
+  def signature(w: Watchable): Long = w match {
     case Watchable.Path(p, quick, sig) =>
       new PathRef(os.Path(p), quick, sig, PathRef.Revalidate.Once).sig
     case Watchable.Value(f, sig, pretty) => sig