diff --git a/src/fs.ts b/src/fs.ts index 459cec1..f9dbdd4 100644 --- a/src/fs.ts +++ b/src/fs.ts @@ -24,8 +24,18 @@ export interface Stat { // browser too. Host devices are never touched. export const DEVICES = new Set(["/dev/null", "/dev/zero", "/dev/random", "/dev/urandom"]); +// Errors carry an errno name in `code`, as Node's do. +export interface FileHandle { + size(): number; + read(dst: Uint8Array, off: number, len: number, pos: number): number; + write(src: Uint8Array, pos: number): void; + truncate(size: number): void; + close(): void; +} + export interface Backend { readonly writable: boolean; + openFile?(path: string, write: boolean, truncate: boolean, readable: boolean): FileHandle; // A host backend can follow symlinks before resolving parent components. resolvePath?(path: string): string | null; stat(path: string): Stat | null; @@ -161,6 +171,25 @@ export class VFS { return r ? r[0].read(r[1]) : null; } + canOpenFile(path: string): boolean { + const r = this.resolve(path); + return !!r && r[0].writable && !!r[0].openFile; + } + + openFile(path: string, write: boolean, truncate: boolean, readable: boolean): FileHandle | undefined { + const r = this.resolve(path); + if (!r || !r[0].writable || !r[0].openFile) return undefined; + const h = r[0].openFile(r[1], write, truncate, readable); + // writes through the handle are changes watchers see, as writes through VFS.write are + return { + size: () => h.size(), + read: (dst, off, len, pos) => h.read(dst, off, len, pos), + close: () => h.close(), + write: (src, pos) => { h.write(src, pos); this.changed(path, "modify"); }, + truncate: (size) => { h.truncate(size); this.changed(path, "modify"); }, + }; + } + // Whether part of a file can be read without reading all of it. canReadRange(path: string): boolean { const r = this.resolve(path); diff --git a/src/natives/io.ts b/src/natives/io.ts index 9e3d3a8..4a6823f 100644 --- a/src/natives/io.ts +++ b/src/natives/io.ts @@ -4,7 +4,8 @@ import { JArray, JObject } from "../runtime.ts"; import { JThread, PendingThrow } from "../thread.ts"; import type { NativeGen } from "../thread.ts"; -import type { Stat, VFS } from "../fs.ts"; +import type { JVM } from "../jvm.ts"; +import type { FileHandle, Stat, VFS } from "../fs.ts"; import { closeSockFd } from "../sockets.ts"; import { nonNull, u8 } from "./util.ts"; import type { Natives } from "./util.ts"; @@ -33,10 +34,53 @@ export function processPipe(t: JThread, fd: number): ProcessPipe | undefined { } // A file's bytes, shared by every descriptor open on it, as an inode is. +// Small files live in data; past STREAM_MIN the file is read and written on +// the host through one handle instead, so its size does not cost memory. +// size is kept here in both cases: changes made outside the VM are not seen. interface Contents { data: Uint8Array; size: number; dirty: boolean; + refs: number; + host?: FileHandle; + hostWritable?: boolean; + hostReadable?: boolean; +} + +// Opens a file's host handle when it outgrows memory. Kept apart from the +// contents, which a boot snapshot saves. +const hostOpeners = new WeakMap FileHandle | undefined>(); +const closesHosts = new WeakSet(); + +// A snapshot cannot hold a host handle, nor the opener a file may still need. +export function streams(f: OpenFile): boolean { + return !!f.contents.host || hostOpeners.has(f.contents); +} + +// Host handles of files the program leaves open close when its VM exits. +function closeHostsOnExit(vm: JVM) { + if (closesHosts.has(vm)) return; + closesHosts.add(vm); + vm.exited.then(() => { + const tab = vm.state.get("fds") as FdTable | undefined; + for (const f of tab?.files.values() ?? []) { + f.contents.host?.close(); + f.contents.host = undefined; + } + }); +} + +// Bigger files stream to the host where the backend can: small ones stay in +// memory, where many small reads and writes cost no system call each. +const STREAM_MIN = 4 << 20; + +function hostCall(call: () => T): T { + try { return call(); } + catch (e) { + if (e instanceof PendingThrow) throw e; + const code = (e as { code?: string }).code ?? "EIO"; + throw new PendingThrow("java/io/IOException", ERRNO[code] ?? "Input/output error"); + } } export class OpenFile { @@ -50,7 +94,8 @@ export class OpenFile { detached?: { path: string; stat: Stat; writable: boolean }; constructor(path: string, data: Uint8Array, writable: boolean, contents?: Contents) { this.path = path; - this.contents = contents ?? { data, size: data.length, dirty: false }; + this.contents = contents ?? { data, size: data.length, dirty: false, refs: 0 }; + this.contents.refs++; this.writable = writable; } @@ -64,12 +109,25 @@ export class OpenFile { read(dst: Uint8Array, off: number, len: number): number { if (this.pos >= this.size) return -1; const n = Math.min(len, this.size - this.pos); + const host = this.contents.host; + if (host && n) { + const got = hostCall(() => host.read(dst, off, n, this.pos)); + this.pos += got; + return got || -1; + } dst.set(this.data.subarray(this.pos, this.pos + n), off); this.pos += n; return n; } write(src: Uint8Array) { + const host = this.contents.host; + if (host) { + hostCall(() => host.write(src, this.pos)); + this.pos += src.length; + if (this.pos > this.size) this.size = this.pos; + return; + } // past the end, the gap reads as zeros (the buffer may hold bytes from before a truncation) if (this.pos > this.size) this.data.fill(0, this.size, Math.min(this.pos, this.data.length)); const end = this.pos + src.length; @@ -82,9 +140,18 @@ export class OpenFile { this.pos = end; if (end > this.size) this.size = end; this.dirty = true; + if (this.size >= STREAM_MIN) this.stream(); } setLength(n: number) { + const host = this.contents.host; + if (host) { + hostCall(() => host.truncate(n)); + this.size = n; + if (this.pos > n) this.pos = n; + return; + } + if (n >= STREAM_MIN && this.stream()) return this.setLength(n); if (n > this.data.length) { const next = new Uint8Array(n); next.set(this.data.subarray(0, this.size)); @@ -94,6 +161,40 @@ export class OpenFile { if (this.pos > n) this.pos = n; this.dirty = true; } + + // Moves the bytes to the host file; on failure they stay in memory and the + // flush that writes them out reports the error. + private stream(): boolean { + const c = this.contents; + const open = hostOpeners.get(c); + if (!open || !this.path) return false; + let host: FileHandle | undefined; + try { + host = open(this.path); + if (!host) throw new Error("no host file"); + host.write(c.data.subarray(0, c.size), 0); + host.truncate(c.size); + } catch { + host?.close(); + hostOpeners.delete(c); + return false; + } + c.host = host; + c.hostWritable = true; + c.data = new Uint8Array(0); + c.dirty = false; + return true; + } + + // a mapping keeps the file open after its descriptor closes + retain() { this.contents.refs++; } + release() { + if (--this.contents.refs === 0 && this.contents.host) { + this.contents.host.close(); + this.contents.host = undefined; + } + } + close() { this.release(); } } // Reads and writes a character device as a stream. @@ -177,7 +278,7 @@ function notFound(path: string, why = "No such file or directory"): never { } // Opens a path and returns its descriptor number. -export function openFile(t: JThread, path: string, write: boolean, truncate: boolean, append: boolean): number { +export function openFile(t: JThread, path: string, write: boolean, truncate: boolean, append: boolean, readable = true): number { const fs = t.vm.fs; const abs = resolvePath(t, path); syncPath(t, abs); @@ -190,6 +291,17 @@ export function openFile(t: JThread, path: string, write: boolean, truncate: boo return tab.next++; } if (write && fs.denied(abs)) notFound(path, "Permission denied"); + if (st && !shared && !truncate && st.size >= STREAM_MIN && fs.canOpenFile(abs)) { + const host = openHost(fs, abs, write, readable); + if (host) { + closeHostsOnExit(t.vm); + const f = new OpenFile(abs, new Uint8Array(0), write, { data: new Uint8Array(0), size: host.size(), dirty: false, refs: 0, host, hostWritable: write, hostReadable: readable }); + if (append) { f.pos = f.size; f.append = true; } + const tab = table(t); + tab.files.set(tab.next, f); + return tab.next++; + } + } let data: Uint8Array; if (!st) { if (!write) notFound(path); @@ -203,7 +315,26 @@ export function openFile(t: JThread, path: string, write: boolean, truncate: boo else if (truncate) data = new Uint8Array(0); else data = fs.read(abs) ?? notFound(path, "Operation not supported"); } + // a descriptor joining a host handle that lacks its access reopens it with both + if (shared?.host && ((write && !shared.hostWritable) || (readable && !shared.hostReadable))) { + const w = write || !!shared.hostWritable, r = readable || !!shared.hostReadable; + const host = openHost(fs, abs, w, r) ?? notFound(path, "Permission denied"); + shared.host.close(); + Object.assign(shared, { host, hostWritable: w, hostReadable: r }); + } const f = new OpenFile(abs, data, write, shared); + // only a writer can grow a file, so read-only files never hold an opener + if (write && !hostOpeners.has(f.contents) && fs.canOpenFile(abs)) { + const vm = t.vm; + const c = f.contents; + hostOpeners.set(c, (current) => { + closeHostsOnExit(vm); + // descriptors already sharing the contents may read through the handle + const host = openHost(fs, current, write, true); + c.hostReadable = !!host; + return host ?? openHost(fs, current, write, false); + }); + } if (truncate && st) f.setLength(0); if (append) { f.pos = f.size; f.append = true; } const tab = table(t); @@ -212,6 +343,11 @@ export function openFile(t: JThread, path: string, write: boolean, truncate: boo return fd; } +function openHost(fs: VFS, abs: string, write: boolean, readable: boolean): FileHandle | undefined { + try { return fs.openFile(abs, write, false, readable); } + catch { return undefined; } +} + // Writes reach the file system when a file is closed. Another open or a stat // of the same path sees them at once on a real OS, so write them out first. export function syncPath(t: JThread, abs: string) { @@ -224,6 +360,13 @@ export function setMtime(t: JThread, abs: string, ms: number, ns?: bigint): bool return t.vm.fs.setMtime(abs, ms, ns); } +// write(2) on a descriptor opened read-only fails with EBADF. +function writableFile(t: JThread, fd: number): OpenFile { + const f = fileOf(t, fd); + if (!f.writable) throw new PendingThrow("java/io/IOException", "Bad file descriptor"); + return f; +} + export function fileOf(t: JThread, fd: number): OpenFile { const f = table(t).files.get(fd); if (!f) throw new PendingThrow("java/io/IOException", "Stream Closed"); @@ -251,7 +394,7 @@ const ERRNO: Record = { ENOSPC: "No space left on device", EFBIG: "File too large", EDQUOT: "Disk quota exceeded", EACCES: "Permission denied", EPERM: "Operation not permitted", EROFS: "Read-only file system", ENOENT: "No such file or directory", ENAMETOOLONG: "File name too long", ENOTDIR: "Not a directory", - EIO: "Input/output error", ELOOP: "Too many levels of symbolic links", + EIO: "Input/output error", ELOOP: "Too many levels of symbolic links", EOPNOTSUPP: "Operation not supported", }; // Where Java learns of a write that failed: HotSpot reports it from write(), @@ -281,6 +424,7 @@ export function closeFd(t: JThread, fd: number) { if (f) { const ok = flush(t, f); tab.files.delete(fd); + f.close(); if (f.writable && f.path) t.vm.opts.fileClosed?.(f.path); // the descriptor is closed either way, as with close(2) if (!ok) throw writeFailed(t, f.path); @@ -354,9 +498,7 @@ function writeOut(t: JThread, fd: number, bytes: Uint8Array) { else if (fd === 2) t.vm.opts.stderr(bytes.slice()); else if (processPipe(t, fd)) return writeProcessPipe(t, processPipe(t, fd)!, bytes); else { - const f = fileOf(t, fd); - if (!f.writable) throw new PendingThrow("java/io/IOException", "Bad file descriptor"); - f.write(bytes); + writableFile(t, fd).write(bytes); } } @@ -484,7 +626,7 @@ export const natives: Natives = { }, [`${FOS}.open0(Ljava/lang/String;Z)V`]: (t, [self, path, append]) => - setFd(t, self, FOS, openFile(t, t.vm.jsString(path), true, !append, append === 1)), + setFd(t, self, FOS, openFile(t, t.vm.jsString(path), true, !append, append === 1, false)), [`${FOS}.writeBytes([BIIZ)V`]: (t, [self, b, off, len]) => { const a = nonNull(b); if (off < 0 || len < 0 || off + len > a.a.length) throw new PendingThrow("java/lang/IndexOutOfBoundsException"); @@ -540,11 +682,11 @@ export const natives: Natives = { if (off < 0 || len < 0 || off + len > a.a.length) throw new PendingThrow("java/lang/IndexOutOfBoundsException"); return len === 0 ? 0 : fileOf(t, fdOf(t, self, RAF)).read(u8(a), off, len); }, - [`${RAF}.write0(I)V`]: (t, [self, b]) => fileOf(t, fdOf(t, self, RAF)).write(Uint8Array.of(b & 0xff)), + [`${RAF}.write0(I)V`]: (t, [self, b]) => writableFile(t, fdOf(t, self, RAF)).write(Uint8Array.of(b & 0xff)), [`${RAF}.writeBytes0([BII)V`]: (t, [self, b, off, len]) => { const a = nonNull(b); if (off < 0 || len < 0 || off + len > a.a.length) throw new PendingThrow("java/lang/IndexOutOfBoundsException"); - fileOf(t, fdOf(t, self, RAF)).write(u8(a, off, len)); + writableFile(t, fdOf(t, self, RAF)).write(u8(a, off, len)); }, [`${RAF}.getFilePointer()J`]: (t, [self]) => BigInt(fileOf(t, fdOf(t, self, RAF)).pos), [`${RAF}.seek0(J)V`]: (t, [self, pos]) => { fileOf(t, fdOf(t, self, RAF)).pos = Number(pos); }, diff --git a/src/natives/nio.ts b/src/natives/nio.ts index de8b8f9..306a16b 100644 --- a/src/natives/nio.ts +++ b/src/natives/nio.ts @@ -152,7 +152,7 @@ export const natives: Natives = { if (st?.dir && write) unixError(EISDIR); if ((write || !st) && t.vm.fs.denied(abs)) unixError(EACCES); try { - const fd = openFile(t, abs, write, (flags & O_TRUNC) !== 0, (flags & O_APPEND) !== 0); + const fd = openFile(t, abs, write, (flags & O_TRUNC) !== 0, (flags & O_APPEND) !== 0, (flags & O_ACCMODE) !== 1); fileOf(t, fd).ino = fileInode(t, abs); return fd; } catch (e) { @@ -355,6 +355,7 @@ function writeAt(t: JThread, fdObj: JObject, addr: bigint, len: number, pos: num return len; } const f = chanFile(t, fdObj); + if (!f.writable) throw new PendingThrow("java/io/IOException", "Bad file descriptor"); const save = f.pos; if (pos !== null) f.pos = pos; f.write(data); @@ -407,7 +408,10 @@ Object.assign(natives, { return BigInt(f.pos); }, [`${UFD}truncate0(Ljava/io/FileDescriptor;J)I`]: (t: JThread, [fd, size]: any[]) => { - chanFile(t, fd).setLength(Number(size)); + const f = chanFile(t, fd); + // ftruncate(2) on a descriptor not open for writing + if (!f.writable) throw new PendingThrow("java/io/IOException", "Invalid argument"); + f.setLength(Number(size)); return 0; }, [`${UFD}size0(Ljava/io/FileDescriptor;)J`]: (t: JThread, [fd]: any[]) => BigInt(chanFile(t, fd).size), @@ -431,12 +435,18 @@ Object.assign(natives, { const n = Math.min(f.size - Number(pos), Number(len)); if (n > 0) f.read(t.vm.mem.u8, addr, n); f.pos = save; - if (prot === MAP_RW) mappings(t).set(addr, { f, pos: Number(pos), len: Number(len) }); + if (prot === MAP_RW) { + f.retain(); + mappings(t).set(addr, { f, pos: Number(pos), len: Number(len) }); + } return BigInt(addr); }, [`${UFD}unmap0(JJ)I`]: (t: JThread, [addr]: any[]) => { const m = mappings(t).get(Number(addr)); - if (m) writeBack(t, m, Number(addr), Number(addr), m.len); + if (m) { + try { writeBack(t, m, Number(addr), Number(addr), m.len); } + finally { m.f.release(); } + } mappings(t).delete(Number(addr)); t.vm.mem.freeBlock(Number(addr)); return 0; diff --git a/src/node.ts b/src/node.ts index 1666ca0..da4109c 100644 --- a/src/node.ts +++ b/src/node.ts @@ -38,6 +38,34 @@ export class NodeBackend implements Backend { private host(p: string) { return this.root === "/" ? p : this.root + p; } + openFile(p: string, write: boolean, truncate: boolean, readable: boolean) { + const flags = write ? (readable ? fs.constants.O_RDWR : fs.constants.O_WRONLY) | fs.constants.O_CREAT | (truncate ? fs.constants.O_TRUNC : 0) : fs.constants.O_RDONLY; + let fd: number; + try { + const host = this.host(p); + const stat = fs.statSync(host, { throwIfNoEntry: false }); + if (stat && !stat.isFile()) throw Object.assign(new Error("Not a regular file"), { code: "EOPNOTSUPP" }); + fd = fs.openSync(host, flags | fs.constants.O_NONBLOCK, 0o666); + if (!fs.fstatSync(fd).isFile()) { + fs.closeSync(fd); + throw Object.assign(new Error("Not a regular file"), { code: "EOPNOTSUPP" }); + } + } + catch (e) { + this.writeFailure = [p, (e as NodeJS.ErrnoException).code ?? "EIO"]; + throw e; + } + return { + size: () => fs.fstatSync(fd).size, + read: (dst: Uint8Array, off: number, len: number, pos: number) => fs.readSync(fd, dst, off, len, pos), + write: (src: Uint8Array, pos: number) => { + let n = 0; + while (n < src.length) n += fs.writeSync(fd, src, n, src.length - n, pos + n); + }, + truncate: (size: number) => fs.ftruncateSync(fd, size), + close: () => fs.closeSync(fd), + }; + } resolvePath(p: string): string | null { if (this.root !== "/") return null; try { diff --git a/src/snapshot.ts b/src/snapshot.ts index c5e982e..fd78a76 100644 --- a/src/snapshot.ts +++ b/src/snapshot.ts @@ -8,6 +8,7 @@ import { parseClass } from "./classfile.ts"; import type { JVM } from "./jvm.ts"; import { platformProperties, vmProperties } from "./natives/misc.ts"; import { restorePending, restoreReference } from "./natives/ref.ts"; +import { streams } from "./natives/io.ts"; import { MAX_HEAP } from "./runtime.ts"; import { JArray, JClass, JField, JMethod, JObject, Monitor } from "./runtime.ts"; import { Frame, JThread, NativeFrame, STATUS_RUNNABLE } from "./thread.ts"; @@ -289,6 +290,7 @@ export function dumpVM(vm: JVM, main: JThread, key: string): SnapshotImage { classCount: vm.classCount, state: [...s].filter(([k, x]) => { if (k === "socks" && x.socks.size) throw new Error("snapshot: a socket is open"); + if (k === "fds" && [...x.files.values()].some(streams)) throw new Error("snapshot: a host file that streams is open"); return !TRANSIENT.has(k) && !k.startsWith("parked:"); }).map(([k, v]) => [k, e.enc(v, `state.${k}`)]), pending: e.enc(refs?.pending ?? [], "refs.pending"), diff --git a/test/java/StreamingFile.java b/test/java/StreamingFile.java new file mode 100644 index 0000000..1815829 --- /dev/null +++ b/test/java/StreamingFile.java @@ -0,0 +1,83 @@ +import java.io.RandomAccessFile; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; +import java.nio.file.Files; +import java.nio.file.Path; + +public class StreamingFile { + public static void main(String[] args) throws Exception { + Path path = Files.createTempFile("streaming-file", ".bin"); + try (RandomAccessFile first = new RandomAccessFile(path.toFile(), "rw"); + RandomAccessFile second = new RandomAccessFile(path.toFile(), "rw")) { + byte[] block = new byte[1024 * 1024]; + for (int i = 0; i < block.length; i++) block[i] = (byte) i; + for (int i = 0; i < 16; i++) first.write(block); + first.seek(20L * 1024 * 1024 + 3); + first.write(123); + System.out.println(second.length()); + second.seek(19L * 1024 * 1024); + System.out.println(second.read() + " " + second.read()); + FileChannel channel = second.getChannel(); + ByteBuffer b = ByteBuffer.allocate(4); + channel.read(b, 20L * 1024 * 1024); + System.out.println(b.get(0) + " " + b.get(3)); + channel.truncate(1024 * 1024 + 1); + first.setLength(2 * 1024 * 1024); + first.seek(1024 * 1024); + System.out.println(first.read() + " " + first.read() + " " + second.length()); + } finally { + Files.deleteIfExists(path); + } + Path big = Files.createTempFile("streaming-file", ".bin"); + try { + try (RandomAccessFile out = new RandomAccessFile(big.toFile(), "rw")) { + out.setLength(6L << 20); + out.seek((6L << 20) - 1); + out.write(7); + } + try (RandomAccessFile again = new RandomAccessFile(big.toFile(), "rw")) { + again.seek((6L << 20) - 1); + System.out.println(again.read() + " " + again.length()); + again.write(9); + again.setLength(1000); + System.out.println(again.length() + " " + Files.size(big)); + } + System.out.println(Files.size(big)); + try (RandomAccessFile ro = new RandomAccessFile(big.toFile(), "r")) { + ro.setLength(0); + } catch (java.io.IOException e) { + System.out.println(e.getMessage()); + } + Files.write(big, new byte[5 << 20]); + try (RandomAccessFile reader = new RandomAccessFile(big.toFile(), "r"); + RandomAccessFile writer = new RandomAccessFile(big.toFile(), "rw")) { + writer.seek(1 << 20); + writer.write(42); + reader.seek(1 << 20); + System.out.println(reader.read() + " " + reader.length()); + } + try (var out = new java.io.FileOutputStream(big.toFile(), true); + var in = new java.io.FileInputStream(big.toFile())) { + out.write(7); + System.out.println(in.skip(5 << 20) + " " + in.read() + " " + in.read()); + } + Path moved = big.resolveSibling(big.getFileName() + ".moved"); + try (var out = new java.io.FileOutputStream(big.toFile())) { + out.write(new byte[1 << 20]); + Files.move(big, moved); + for (int i = 0; i < 5; i++) out.write(new byte[1 << 20]); + } + System.out.println(Files.exists(big) + " " + Files.size(moved)); + Files.delete(moved); + Files.write(big, new byte[1000]); + try (RandomAccessFile ro = new RandomAccessFile(big.toFile(), "r")) { + ro.write(1); + System.out.println("wrote"); + } catch (java.io.IOException e) { + System.out.println(e.getMessage() + " " + Files.size(big)); + } + } finally { + Files.deleteIfExists(big); + } + } +} diff --git a/test/java/WatchService.java b/test/java/WatchService.java index 63ec522..ac8f0dd 100644 --- a/test/java/WatchService.java +++ b/test/java/WatchService.java @@ -26,6 +26,13 @@ public static void main(String[] args) throws Exception { Files.writeString(file, "first"); Files.writeString(file, "second"); Files.delete(file); + // past 4 MiB the file streams to the host; its writes still report changes + Path big = dir.resolve("big"); + try (var out = Files.newOutputStream(big)) { + byte[] block = new byte[1 << 20]; + for (int i = 0; i < 6; i++) out.write(block); + } + Files.delete(big); Thread.sleep(100); collect(watch, events); } diff --git a/test/jtreg/passing.txt b/test/jtreg/passing.txt index 6e20b47..1266d2a 100644 --- a/test/jtreg/passing.txt +++ b/test/jtreg/passing.txt @@ -2127,6 +2127,7 @@ java/nio/file/Files/CopyToNonDefaultFS.java java/nio/file/Files/CreateDirectories.java java/nio/file/Files/CustomOptions.java java/nio/file/Files/FileAttributes.java +java/nio/file/Files/InterruptCopy.java java/nio/file/Files/Links.java java/nio/file/Files/Misc.java java/nio/file/Files/Mismatch.java