Implement WatchService for changes made inside the VM

docs/support.md

@@ -17,6 +17,9 @@ java-web targets headless Java: console apps, libraries and javac.
   java-web VM on the same image. A child cannot use a directory that exists only
   in the VM, such as `/jdk`, as its working directory. `ProcessHandle` reads
   other processes from Linux's `/proc`; elsewhere it sees only its own children.
+- `WatchService` reports files the program itself, or another thread in the
+  same VM, creates, changes or deletes. It does not see changes made outside
+  the VM, including by child processes the program starts.
 - Not supported: GUI apps (AWT and Swing windows), `ConvolveOp`, child
   processes in the browser, and native libraries other than JS natives.
 

src/fs.ts

@@ -70,6 +70,17 @@ export class VFS {
   private setTimes = new Map<string, { ms: number; ns: bigint }>();
   private mounts: { prefix: string; backend: Backend }[] = [];
   private hidden: string[] = [];
+  private watchers = new Set<(path: string, event: "create" | "modify" | "delete") => void>();
+
+  watchChanges(watcher: (path: string, event: "create" | "modify" | "delete") => void): () => void {
+    this.watchers.add(watcher);
+    return () => this.watchers.delete(watcher);
+  }
+
+  private changed(path: string, event: "create" | "modify" | "delete") {
+    path = normalize(path);
+    for (const watcher of this.watchers) watcher(path, event);
+  }
 
   resolvePath(path: string): string {
     const normalized = normalize(path);
@@ -174,7 +185,11 @@ export class VFS {
 
   write(path: string, data: Uint8Array): boolean {
     const r = this.resolve(path);
-    return !!r && r[0].writable && !!r[0].write?.(r[1], data);
+    if (!r || !r[0].writable) return false;
+    const exists = !!r[0].stat(r[1]);
+    if (!r[0].write?.(r[1], data)) return false;
+    this.changed(path, exists ? "modify" : "create");
+    return true;
   }
 
   writeError(path: string): string | undefined {
@@ -184,19 +199,24 @@ export class VFS {
 
   mkdir(path: string): boolean {
     const r = this.resolve(path);
-    return !!r && r[0].writable && !!r[0].mkdir?.(r[1]);
+    if (!r || !r[0].writable || !r[0].mkdir?.(r[1])) return false;
+    this.changed(path, "create");
+    return true;
   }
 
   remove(path: string): boolean {
     const r = this.resolve(path);
     if (!r || !r[0].writable || !r[0].remove?.(r[1])) return false;
     this.setTimes.delete(normalize(path));
+    this.changed(path, "delete");
     return true;
   }
 
   rename(from: string, to: string): boolean {
     const a = this.resolve(from), b = this.resolve(to);
     if (!a || !b || a[0] !== b[0] || !a[0].writable || !a[0].rename?.(a[1], b[1])) return false;
+    this.changed(from, "delete");
+    this.changed(to, "create");
     // set times move with the files, which keep their times
     const src = normalize(from), dst = normalize(to);
     const under = (p: string, root: string) => p === root || p.startsWith(root + "/");
@@ -224,6 +244,7 @@ export class VFS {
   setMtime(path: string, ms: number, ns?: bigint): boolean {
     const r = this.resolve(path);
     if (!r || !r[0].writable || !r[0].setMtime?.(r[1], ms)) return false;
+    this.changed(path, "modify");
     const n = normalize(path);
     const st = ns === undefined ? null : r[0].stat(r[1]);
     // only a time the host kept, give or take its rounding; a host can clamp it

src/natives/nio.ts

@@ -4,7 +4,7 @@
 import type { JObject } from "../runtime.ts";
 import type { Stat } from "../fs.ts";
 import { preCloseSock } from "../sockets.ts";
-import { sockRead, sockReadv, sockWrite, sockWritev } from "./sockets.ts";
+import { closeInotify, sockRead, sockReadv, sockWrite, sockWritev } from "./sockets.ts";
 import { JThread, PendingThrow } from "../thread.ts";
 import { closeFd, endsInDot, fileInode, fileOf, flush, flushOrThrow, openFile, removePath, renamePath, resolvePath, setMtime, syncPath, writeFailed } from "./io.ts";
 import type { OpenFile } from "./io.ts";
@@ -164,7 +164,10 @@ export const natives: Natives = {
       throw e;
     }
   },
-  [`${N}close0(I)V`]: (t, [fd]) => closeFd(t, fd),
+  [`${N}close0(I)V`]: (t, [fd]) => {
+    closeInotify(t, fd);
+    closeFd(t, fd);
+  },
   [`${N}dup(I)I`]: () => unixError(ENOSYS),
   [`${N}mkdir0(JI)V`]: (t, [p]) => {
     const abs = path(t, p);

src/natives/sockets.ts

@@ -16,7 +16,77 @@ import type { Natives } from "./util.ts";
 
 const FD = "java/io/FileDescriptor";
 const IOS_EOF = -1, IOS_UNAVAILABLE = -2, IOS_INTERRUPTED = -3;
-const EEXIST = 17, EBADF = 9, EPROTO = 71;
+const EEXIST = 17, EBADF = 9, EPROTO = 71, ENOTDIR = 20, EFAULT = 14, ENAMETOOLONG = 36;
+
+type Inotify = { sock: Sock; watches: Map<number, { path: string; mask: number }>; paths: Map<string, number>; nextWd: number; overflowed: boolean; remove: () => void };
+const IN_MODIFY = 0x2, IN_ATTRIB = 0x4, IN_MOVED_FROM = 0x40, IN_MOVED_TO = 0x80;
+const IN_CREATE = 0x100, IN_DELETE = 0x200, IN_Q_OVERFLOW = 0x4000;
+const IN_IGNORED = 0x8000;
+const WATCH_EVENTS = IN_MODIFY | IN_ATTRIB | IN_MOVED_FROM | IN_MOVED_TO | IN_CREATE | IN_DELETE;
+const MAX_INOTIFY_BYTES = 1 << 20;
+
+function inotify(t: JThread, fd: number): Inotify | undefined {
+  return (t.vm.state.get("inotify") as Map<number, Inotify> | undefined)?.get(fd);
+}
+
+function initInotify(t: JThread): number {
+  let all = t.vm.state.get("inotify") as Map<number, Inotify> | undefined;
+  if (!all) t.vm.state.set("inotify", (all = new Map()));
+  const sock = new Sock("stream");
+  sock.connected = true;
+  sock.blocking = false;
+  const state: Inotify = { sock, watches: new Map(), paths: new Map(), nextWd: 1, overflowed: false, remove: () => {} };
+  state.remove = t.vm.fs.watchChanges((path, event) => {
+    const slash = path.lastIndexOf("/");
+    const parent = slash === 0 ? "/" : path.slice(0, slash);
+    const name = new TextEncoder().encode(path.slice(slash + 1));
+    for (const [wd, watch] of state.watches) {
+      if (event === "delete" && watch.path === path) {
+        const bytes = new Uint8Array(16);
+        const view = new DataView(bytes.buffer);
+        view.setInt32(0, wd, true);
+        view.setUint32(4, IN_IGNORED, true);
+        sock.push(bytes);
+        state.watches.delete(wd);
+        state.paths.delete(path);
+        continue;
+      }
+      if (watch.path !== parent) continue;
+      const mask = event === "create" ? IN_CREATE : event === "delete" ? IN_DELETE : IN_MODIFY;
+      if (!(watch.mask & mask)) continue;
+      const len = (name.length + 1 + 3) & ~3;
+      const bytes = new Uint8Array(16 + len);
+      const view = new DataView(bytes.buffer);
+      view.setInt32(0, wd, true);
+      view.setUint32(4, mask, true);
+      view.setUint32(8, 0, true);
+      view.setUint32(12, len, true);
+      bytes.set(name, 16);
+      // as Linux: a full queue keeps its events, gains one overflow record and drops new ones
+      if (sock.rxBytes + bytes.length > MAX_INOTIFY_BYTES) {
+        if (!state.overflowed) {
+          const overflow = new Uint8Array(16);
+          new DataView(overflow.buffer).setUint32(4, IN_Q_OVERFLOW, true);
+          sock.push(overflow);
+          state.overflowed = true;
+        }
+        continue;
+      }
+      state.overflowed = false;
+      sock.push(bytes);
+    }
+  });
+  const fd = register(t, sock);
+  all.set(fd, state);
+  t.vm.exited.then(() => state.remove());
+  return fd;
+}
+
+// Stops an inotify descriptor's watches before the descriptor closes.
+export function closeInotify(t: JThread, fd: number) {
+  inotify(t, fd)?.remove();
+  (t.vm.state.get("inotify") as Map<number, Inotify> | undefined)?.delete(fd);
+}
 
 // the JDK thread token for signalling a thread out of a blocking call
 const blocked = new WeakMap<JThread, () => void>();
@@ -143,6 +213,37 @@ function* pollSock(t: JThread, s: Sock, events: number, timeout: number): Native
   }
 }
 
+function* pollPair(t: JThread, a: Sock, b: Sock): NativeGen {
+  for (;;) {
+    const ready = ((a.poll() & POLLIN) ? 1 : 0) + ((b.poll() & POLLIN) ? 1 : 0);
+    if (ready) return ready;
+    yield t.block(new Promise<void>((resolve) => {
+      let done = false;
+      const wake = () => {
+        if (done) return;
+        done = true;
+        a.waiters.delete(wake);
+        b.waiters.delete(wake);
+        blocked.delete(t);
+        resolve();
+      };
+      a.waiters.add(wake);
+      b.waiters.add(wake);
+      blocked.set(t, wake);
+      if ((a.poll() & POLLIN) || (b.poll() & POLLIN)) wake();
+    }));
+  }
+}
+
+function nativePath(t: JThread, address: bigint): string {
+  const bytes = t.vm.mem.u8;
+  const start = Number(address);
+  if (start <= 0 || start >= bytes.length) throw ioError(EFAULT);
+  const len = bytes.subarray(start, start + 4096).indexOf(0);
+  if (len < 0) throw ioError(ENAMETOOLONG);
+  return resolvePath(t, new TextDecoder().decode(bytes.subarray(start, start + len)));
+}
+
 // Takes stream bytes: into(cap) moves up to cap bytes and returns the count.
 function* recv(t: JThread, s: Sock, len: number, into: (cap: number) => number): NativeGen {
   if (len === 0) return 0;
@@ -434,6 +535,65 @@ const DC = "sun/nio/ch/DatagramChannelImpl.";
 const NSA = "sun/nio/ch/NativeSocketAddress.";
 
 export const natives: Natives = {
+  "sun/nio/fs/LinuxWatchService.eventSize()I": () => 16,
+  "sun/nio/fs/LinuxWatchService.eventOffsets()[I": (t) => {
+    const a = t.vm.newArray(t.vm.primClass("I"), 5);
+    (a.a as Int32Array).set([0, 4, 8, 12, 16]);
+    return a;
+  },
+  "sun/nio/fs/LinuxWatchService.inotifyInit()I": (t) => initInotify(t),
+  "sun/nio/fs/LinuxWatchService.inotifyAddWatch(IJI)I": (t, [fd, address, mask]) => {
+    const state = inotify(t, fd);
+    if (!state || !socksOf(t.vm.state).has(fd)) throw ioError(EBADF);
+    const path = nativePath(t, address);
+    const st = t.vm.fs.stat(path);
+    if (!st) throw ioError(ENOENT);
+    if (!st.dir) throw ioError(ENOTDIR);
+    let wd = state.paths.get(path);
+    if (wd === undefined) {
+      wd = state.nextWd++;
+      state.paths.set(path, wd);
+    }
+    state.watches.set(wd, { path, mask: mask & WATCH_EVENTS });
+    return wd;
+  },
+  "sun/nio/fs/LinuxWatchService.inotifyRmWatch(II)V": (t, [fd, wd]) => {
+    const state = inotify(t, fd);
+    if (!state) throw ioError(EBADF);
+    const watch = state.watches.get(wd);
+    if (watch) {
+      state.watches.delete(wd);
+      state.paths.delete(watch.path);
+    }
+  },
+  "sun/nio/fs/LinuxWatchService.configureBlocking(IZ)V": (t, [fd, blocking]) => {
+    const s = socksOf(t.vm.state).get(fd);
+    if (!s) throw ioError(EBADF);
+    s.blocking = blocking === 1;
+  },
+  "sun/nio/fs/LinuxWatchService.socketpair([I)V": (t, [sv]) => {
+    const [r, w] = newPipe();
+    r.blocking = w.blocking = false;
+    sv.a[0] = register(t, r);
+    sv.a[1] = register(t, w);
+  },
+  "sun/nio/fs/LinuxWatchService.poll(II)I": function* (t, [fd1, fd2]): NativeGen {
+    const a = socksOf(t.vm.state).get(fd1), b = socksOf(t.vm.state).get(fd2);
+    if (!a || !b) throw ioError(EBADF);
+    return yield* pollPair(t, a, b);
+  },
+  "sun/nio/fs/UnixNativeDispatcher.read0(IJI)I": function* (t, [fd, address, len]): NativeGen {
+    const s = socksOf(t.vm.state).get(fd);
+    if (!s) throw new PendingThrow("sun/nio/fs/UnixException", null, { desc: "(I)V", args: [EBADF] });
+    const n = yield* recv(t, s, len, (count) => s.take(t.vm.mem.u8, Number(address), count));
+    if (n === IOS_UNAVAILABLE) throw new PendingThrow("sun/nio/fs/UnixException", null, { desc: "(I)V", args: [11] });
+    return n;
+  },
+  "sun/nio/fs/UnixNativeDispatcher.write0(IJI)I": function* (t, [fd, address, len]): NativeGen {
+    const s = socksOf(t.vm.state).get(fd);
+    if (!s) throw new PendingThrow("sun/nio/fs/UnixException", null, { desc: "(I)V", args: [EBADF] });
+    return yield* xmit(t, s, t.vm.mem.u8.slice(Number(address), Number(address) + len));
+  },
   [`${N}isIPv6Available0()Z`]: () => 0,
   [`${N}isReusePortAvailable0()Z`]: () => 1,
   [`${N}isExclusiveBindAvailable()I`]: () => -1,

test/java/WatchService.java

@@ -0,0 +1,35 @@
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardWatchEventKinds;
+import java.nio.file.WatchKey;
+import java.util.TreeSet;
+
+public class WatchService {
+  private static void collect(java.nio.file.WatchService watch, TreeSet<String> events) throws Exception {
+    WatchKey key;
+    while ((key = watch.poll()) != null) {
+      for (var event : key.pollEvents()) {
+        if (event.kind() != StandardWatchEventKinds.OVERFLOW)
+          events.add(event.kind().name() + " " + event.context());
+      }
+      key.reset();
+    }
+  }
+
+  public static void main(String[] args) throws Exception {
+    Path dir = Files.createTempDirectory(Path.of("."), "watch");
+    Path file = dir.resolve("item");
+    TreeSet<String> events = new TreeSet<>();
+    try (java.nio.file.WatchService watch = dir.getFileSystem().newWatchService()) {
+      dir.register(watch, StandardWatchEventKinds.ENTRY_CREATE,
+          StandardWatchEventKinds.ENTRY_MODIFY, StandardWatchEventKinds.ENTRY_DELETE);
+      Files.writeString(file, "first");
+      Files.writeString(file, "second");
+      Files.delete(file);
+      Thread.sleep(100);
+      collect(watch, events);
+    }
+    for (String event : events) System.out.println(event);
+    Files.delete(dir);
+  }
+}

test/jtreg/passing.txt

@@ -2143,6 +2143,14 @@ java/nio/file/Path/PathOps.java
 java/nio/file/Path/ToRealPath.java
 java/nio/file/Path/UriImportExport.java
 java/nio/file/PathMatcher/Basic.java
+java/nio/file/WatchService/Basic.java
+java/nio/file/WatchService/LotsOfCancels.java
+java/nio/file/WatchService/LotsOfCloses.java
+java/nio/file/WatchService/LotsOfEntries.java
+java/nio/file/WatchService/LotsOfEvents.java
+java/nio/file/WatchService/MayFlies.java
+java/nio/file/WatchService/SensitivityModifier.java
+java/nio/file/WatchService/UpdateInterference.java
 java/nio/file/attribute/AclEntry/EmptySet.java
 java/nio/file/attribute/AclFileAttributeView/Basic.java
 java/nio/file/attribute/BasicFileAttributeView/Basic.java