From 069b42a7cc1275969859dc60e7c303528ca2dccb Mon Sep 17 00:00:00 2001 From: Ciro Spaciari Date: Sat, 24 Jun 2023 03:24:34 -0300 Subject: [feat] fs.watch (#3249) * initial support * add types * fix comment * fix types * bigfix up * more fixes * fix some encoding support for watch * fix rename event * fixup * fix latin1 * add fs_events, still failing some tests * fixuup * remove unecesary check * readd tests ops * this is necessary? just testing CI/CD weird errors * just use dupe here * cleanup and fix deinit * fix zig upgrade --- src/bun.js/node/node_fs_watcher.zig | 913 ++++++++++++++++++++++++++++++++++++ 1 file changed, 913 insertions(+) create mode 100644 src/bun.js/node/node_fs_watcher.zig (limited to 'src/bun.js/node/node_fs_watcher.zig') diff --git a/src/bun.js/node/node_fs_watcher.zig b/src/bun.js/node/node_fs_watcher.zig new file mode 100644 index 000000000..397d51916 --- /dev/null +++ b/src/bun.js/node/node_fs_watcher.zig @@ -0,0 +1,913 @@ +const std = @import("std"); +const JSC = @import("root").bun.JSC; +const bun = @import("root").bun; +const Fs = @import("../../fs.zig"); +const Path = @import("../../resolver/resolve_path.zig"); +const Encoder = JSC.WebCore.Encoder; + +const FSEvents = @import("./fs_events.zig"); + +const VirtualMachine = JSC.VirtualMachine; +const EventLoop = JSC.EventLoop; +const PathLike = JSC.Node.PathLike; +const ArgumentsSlice = JSC.Node.ArgumentsSlice; +const Output = bun.Output; +const string = bun.string; +const StoredFileDescriptorType = bun.StoredFileDescriptorType; +const Environment = bun.Environment; + +pub const FSWatcher = struct { + const watcher = @import("../../watcher.zig"); + const options = @import("../../options.zig"); + pub const Watcher = watcher.NewWatcher(*FSWatcher); + const log = Output.scoped(.FSWatcher, false); + + pub const ChangeEvent = struct { + hash: Watcher.HashType = 0, + event_type: FSWatchTask.EventType = .change, + time_stamp: i64 = 0, + }; + + onAccept: std.ArrayHashMapUnmanaged(FSWatcher.Watcher.HashType, bun.BabyList(OnAcceptCallback), bun.ArrayIdentityContext, false) = .{}, + ctx: *VirtualMachine, + js_watcher: ?*JSObject = null, + watcher_instance: ?*FSWatcher.Watcher = null, + verbose: bool = false, + file_paths: bun.BabyList(string) = .{}, + entry_path: ?string = null, + entry_dir: string = "", + last_change_event: ChangeEvent = .{}, + + pub fn toJS(this: *FSWatcher) JSC.JSValue { + return if (this.js_watcher) |js| js.js_this else JSC.JSValue.jsUndefined(); + } + + pub fn eventLoop(this: FSWatcher) *EventLoop { + return this.ctx.eventLoop(); + } + + pub fn enqueueTaskConcurrent(this: FSWatcher, task: *JSC.ConcurrentTask) void { + this.eventLoop().enqueueTaskConcurrent(task); + } + + pub fn deinit(this: *FSWatcher) void { + while (this.file_paths.popOrNull()) |file_path| { + bun.default_allocator.destroy(file_path); + } + this.file_paths.deinitWithAllocator(bun.default_allocator); + if (this.entry_path) |path| { + this.entry_path = null; + bun.default_allocator.destroy(path); + } + bun.default_allocator.destroy(this); + } + + pub const FSWatchTask = struct { + ctx: *FSWatcher, + count: u8 = 0, + + entries: [8]Entry = undefined, + concurrent_task: JSC.ConcurrentTask = undefined, + + pub const EventType = enum { + rename, + change, + @"error", + abort, + }; + + pub const EventFreeType = enum { + destroy, + free, + none, + }; + + pub const Entry = struct { + file_path: string, + event_type: EventType, + free_type: EventFreeType, + }; + + pub fn append(this: *FSWatchTask, file_path: string, event_type: EventType, free_type: EventFreeType) void { + if (this.count == 8) { + this.enqueue(); + var ctx = this.ctx; + this.* = .{ + .ctx = ctx, + .count = 0, + }; + } + + this.entries[this.count] = .{ + .file_path = file_path, + .event_type = event_type, + .free_type = free_type, + }; + this.count += 1; + } + + pub fn run(this: *FSWatchTask) void { + // this runs on JS Context + if (this.ctx.js_watcher) |js_watcher| { + for (this.entries[0..this.count]) |entry| { + switch (entry.event_type) { + .rename => { + js_watcher.emit(entry.file_path, "rename"); + }, + .change => { + js_watcher.emit(entry.file_path, "change"); + }, + .@"error" => { + // file_path is the error message in this case + js_watcher.emitError(entry.file_path); + }, + .abort => { + js_watcher.emitIfAborted(); + }, + } + } + } + } + + pub fn enqueue(this: *FSWatchTask) void { + if (this.count == 0) + return; + + var that = bun.default_allocator.create(FSWatchTask) catch unreachable; + + that.* = this.*; + this.count = 0; + that.concurrent_task.task = JSC.Task.init(that); + this.ctx.enqueueTaskConcurrent(&that.concurrent_task); + } + + pub fn deinit(this: *FSWatchTask) void { + while (this.count > 0) { + this.count -= 1; + switch (this.entries[this.count].free_type) { + .destroy => bun.default_allocator.destroy(this.entries[this.count].file_path), + .free => bun.default_allocator.free(this.entries[this.count].file_path), + else => {}, + } + } + bun.default_allocator.destroy(this); + } + }; + + fn NewCallback(comptime FunctionSignature: type) type { + return union(enum) { + javascript_callback: JSC.Strong, + zig_callback: struct { + ptr: *anyopaque, + function: *const FunctionSignature, + }, + }; + } + + pub const OnAcceptCallback = NewCallback(fn ( + vm: *JSC.VirtualMachine, + specifier: []const u8, + ) void); + + fn addDirectory(ctx: *FSWatcher, fs_watcher: *FSWatcher.Watcher, fd: StoredFileDescriptorType, file_path: string, recursive: bool, buf: *[bun.MAX_PATH_BYTES + 1]u8, is_entry_path: bool) !void { + var dir_path_clone = bun.default_allocator.dupeZ(u8, file_path) catch unreachable; + + if (is_entry_path) { + ctx.entry_path = dir_path_clone; + ctx.entry_dir = dir_path_clone; + } else { + ctx.file_paths.push(bun.default_allocator, dir_path_clone) catch unreachable; + } + fs_watcher.addDirectory(fd, dir_path_clone, FSWatcher.Watcher.getHash(file_path), false) catch |err| { + ctx.deinit(); + fs_watcher.deinit(true); + return err; + }; + + var iter = (std.fs.IterableDir{ .dir = std.fs.Dir{ + .fd = fd, + } }).iterate(); + + while (iter.next() catch |err| { + ctx.deinit(); + fs_watcher.deinit(true); + return err; + }) |entry| { + var parts = [2]string{ dir_path_clone, entry.name }; + var entry_path = Path.joinAbsStringBuf( + Fs.FileSystem.instance.topLevelDirWithoutTrailingSlash(), + buf, + &parts, + .auto, + ); + + buf[entry_path.len] = 0; + var entry_path_z = buf[0..entry_path.len :0]; + + var fs_info = fdFromAbsolutePathZ(entry_path_z) catch |err| { + ctx.deinit(); + fs_watcher.deinit(true); + return err; + }; + + if (fs_info.is_file) { + const file_path_clone = bun.default_allocator.dupeZ(u8, entry_path) catch unreachable; + + ctx.file_paths.push(bun.default_allocator, file_path_clone) catch unreachable; + + fs_watcher.addFile(fs_info.fd, file_path_clone, FSWatcher.Watcher.getHash(entry_path), options.Loader.file, 0, null, false) catch |err| { + ctx.deinit(); + fs_watcher.deinit(true); + return err; + }; + } else { + if (recursive) { + addDirectory(ctx, fs_watcher, fs_info.fd, entry_path, recursive, buf, false) catch |err| { + ctx.deinit(); + fs_watcher.deinit(true); + return err; + }; + } + } + } + } + + pub fn onError( + this: *FSWatcher, + err: anyerror, + ) void { + var current_task: FSWatchTask = .{ + .ctx = this, + }; + current_task.append(@errorName(err), .@"error", .none); + current_task.enqueue(); + } + + pub fn onFSEventUpdate( + ctx: ?*anyopaque, + path: string, + _: bool, + is_rename: bool, + ) void { + const this = bun.cast(*FSWatcher, ctx.?); + + var current_task: FSWatchTask = .{ + .ctx = this, + }; + defer current_task.enqueue(); + + const relative_path = bun.default_allocator.dupe(u8, path) catch unreachable; + const event_type: FSWatchTask.EventType = if (is_rename) .rename else .change; + + current_task.append(relative_path, event_type, .destroy); + } + + pub fn onFileUpdate( + this: *FSWatcher, + events: []watcher.WatchEvent, + changed_files: []?[:0]u8, + watchlist: watcher.Watchlist, + ) void { + var slice = watchlist.slice(); + const file_paths = slice.items(.file_path); + + var counts = slice.items(.count); + const kinds = slice.items(.kind); + var _on_file_update_path_buf: [bun.MAX_PATH_BYTES]u8 = undefined; + + var ctx = this.watcher_instance.?; + defer ctx.flushEvictions(); + defer Output.flush(); + + var bundler = if (@TypeOf(this.ctx.bundler) == *bun.Bundler) + this.ctx.bundler + else + &this.ctx.bundler; + + var fs: *Fs.FileSystem = bundler.fs; + + var current_task: FSWatchTask = .{ + .ctx = this, + }; + defer current_task.enqueue(); + + const time_stamp = std.time.milliTimestamp(); + const time_diff = time_stamp - this.last_change_event.time_stamp; + + for (events) |event| { + const file_path = file_paths[event.index]; + const update_count = counts[event.index] + 1; + counts[event.index] = update_count; + const kind = kinds[event.index]; + + if (comptime Environment.isDebug) { + if (this.verbose) { + Output.prettyErrorln("[watch] {s} ({s}, {})", .{ file_path, @tagName(kind), event.op }); + } + } + + switch (kind) { + .file => { + if (event.op.delete) { + ctx.removeAtIndex( + event.index, + 0, + &.{}, + .file, + ); + } + + var file_hash: FSWatcher.Watcher.HashType = FSWatcher.Watcher.getHash(file_path); + + if (event.op.write or event.op.delete or event.op.rename) { + const event_type: FSWatchTask.EventType = if (event.op.delete or event.op.rename or event.op.move_to) .rename else .change; + // skip consecutive duplicates + if ((this.last_change_event.time_stamp == 0 or time_diff > 1) or this.last_change_event.event_type != event_type and this.last_change_event.hash != file_hash) { + this.last_change_event.time_stamp = time_stamp; + this.last_change_event.event_type = event_type; + this.last_change_event.hash = file_hash; + + const relative_slice = fs.relative(this.entry_dir, file_path); + + if (this.verbose) + Output.prettyErrorln("File changed: {s}", .{relative_slice}); + + const relative_path = bun.default_allocator.dupe(u8, relative_slice) catch unreachable; + + current_task.append(relative_path, event_type, .destroy); + } + } + }, + .directory => { + // macOS should use FSEvents for directories + if (comptime Environment.isMac) { + @panic("Unexpected directory watch"); + } + + const affected = event.names(changed_files); + + for (affected) |changed_name_| { + const changed_name: []const u8 = bun.asByteSlice(changed_name_.?); + if (changed_name.len == 0 or changed_name[0] == '~' or changed_name[0] == '.') continue; + + var file_hash: FSWatcher.Watcher.HashType = 0; + const relative_slice: string = brk: { + var file_path_without_trailing_slash = std.mem.trimRight(u8, file_path, std.fs.path.sep_str); + + @memcpy(_on_file_update_path_buf[0..file_path_without_trailing_slash.len], file_path_without_trailing_slash); + + _on_file_update_path_buf[file_path_without_trailing_slash.len] = std.fs.path.sep; + + @memcpy(_on_file_update_path_buf[file_path_without_trailing_slash.len + 1 ..][0..changed_name.len], changed_name); + const path_slice = _on_file_update_path_buf[0 .. file_path_without_trailing_slash.len + changed_name.len + 1]; + file_hash = FSWatcher.Watcher.getHash(path_slice); + + const relative = fs.relative(this.entry_dir, path_slice); + + break :brk relative; + }; + + // skip consecutive duplicates + const event_type: FSWatchTask.EventType = .rename; // renaming folders, creating folder or files will be always be rename + if ((this.last_change_event.time_stamp == 0 or time_diff > 1) or this.last_change_event.event_type != event_type and this.last_change_event.hash != file_hash) { + const relative_path = bun.default_allocator.dupe(u8, relative_slice) catch unreachable; + + this.last_change_event.time_stamp = time_stamp; + this.last_change_event.event_type = event_type; + this.last_change_event.hash = file_hash; + + current_task.append(relative_path, event_type, .destroy); + + if (this.verbose) + Output.prettyErrorln(" Dir change: {s}", .{relative_path}); + } + } + + if (this.verbose and affected.len == 0) { + Output.prettyErrorln(" Dir change: {s}", .{fs.relative(this.entry_dir, file_path)}); + } + }, + } + } + } + + pub const Arguments = struct { + path: PathLike, + listener: JSC.JSValue, + global_this: JSC.C.JSContextRef, + signal: ?*JSC.AbortSignal, + persistent: bool, + recursive: bool, + encoding: JSC.Node.Encoding, + verbose: bool, + pub fn fromJS(ctx: JSC.C.JSContextRef, arguments: *ArgumentsSlice, exception: JSC.C.ExceptionRef) ?Arguments { + const vm = ctx.vm(); + const path = PathLike.fromJS(ctx, arguments, exception) orelse { + if (exception.* == null) { + JSC.throwInvalidArguments( + "filename must be a string or TypedArray", + .{}, + ctx, + exception, + ); + } + return null; + }; + + if (exception.* != null) return null; + var listener: JSC.JSValue = .zero; + var signal: ?*JSC.AbortSignal = null; + var persistent: bool = true; + var recursive: bool = false; + var encoding: JSC.Node.Encoding = .utf8; + var verbose = false; + if (arguments.nextEat()) |options_or_callable| { + + // options + if (options_or_callable.isObject()) { + if (options_or_callable.get(ctx, "persistent")) |persistent_| { + if (!persistent_.isBoolean()) { + JSC.throwInvalidArguments( + "persistent must be a boolean.", + .{}, + ctx, + exception, + ); + return null; + } + persistent = persistent_.toBoolean(); + } + + if (options_or_callable.get(ctx, "verbose")) |verbose_| { + if (!verbose_.isBoolean()) { + JSC.throwInvalidArguments( + "verbose must be a boolean.", + .{}, + ctx, + exception, + ); + return null; + } + verbose = verbose_.toBoolean(); + } + + if (options_or_callable.get(ctx, "encoding")) |encoding_| { + if (!encoding_.isString()) { + JSC.throwInvalidArguments( + "encoding must be a string.", + .{}, + ctx, + exception, + ); + return null; + } + if (JSC.Node.Encoding.fromJS(encoding_, ctx.ptr())) |node_encoding| { + encoding = node_encoding; + } else { + JSC.throwInvalidArguments( + "invalid encoding.", + .{}, + ctx, + exception, + ); + return null; + } + } + + if (options_or_callable.get(ctx, "recursive")) |recursive_| { + if (!recursive_.isBoolean()) { + JSC.throwInvalidArguments( + "recursive must be a boolean.", + .{}, + ctx, + exception, + ); + return null; + } + recursive = recursive_.toBoolean(); + } + + // abort signal + if (options_or_callable.get(ctx, "signal")) |signal_| { + if (JSC.AbortSignal.fromJS(signal_)) |signal_obj| { + //Keep it alive + signal_.ensureStillAlive(); + signal = signal_obj; + } else { + JSC.throwInvalidArguments( + "signal is not of type AbortSignal.", + .{}, + ctx, + exception, + ); + + return null; + } + } + + // listener + if (arguments.nextEat()) |callable| { + if (!callable.isCell() or !callable.isCallable(vm)) { + exception.* = JSC.toInvalidArguments("Expected \"listener\" callback to be a function", .{}, ctx).asObjectRef(); + return null; + } + listener = callable; + } + } else { + if (!options_or_callable.isCell() or !options_or_callable.isCallable(vm)) { + exception.* = JSC.toInvalidArguments("Expected \"listener\" callback to be a function", .{}, ctx).asObjectRef(); + return null; + } + listener = options_or_callable; + } + } + if (listener == .zero) { + exception.* = JSC.toInvalidArguments("Expected \"listener\" callback", .{}, ctx).asObjectRef(); + return null; + } + + return Arguments{ + .path = path, + .listener = listener, + .global_this = ctx, + .signal = signal, + .persistent = persistent, + .recursive = recursive, + .encoding = encoding, + .verbose = verbose, + }; + } + + pub fn createFSWatcher(this: Arguments) !JSC.JSValue { + const obj = try FSWatcher.init(this); + return obj.toJS(); + } + }; + + pub const JSObject = struct { + signal: ?*JSC.AbortSignal, + persistent: bool, + manager: ?*FSWatcher.Watcher, + fsevents_watcher: ?*FSEvents.FSEventsWatcher, + poll_ref: JSC.PollRef = .{}, + globalThis: ?*JSC.JSGlobalObject, + js_this: JSC.JSValue, + encoding: JSC.Node.Encoding, + closed: bool, + + pub usingnamespace JSC.Codegen.JSFSWatcher; + + pub fn getFSWatcher(this: *JSObject) *FSWatcher { + if (this.manager) |manager| return manager.ctx; + if (this.fsevents_watcher) |manager| return bun.cast(*FSWatcher, manager.ctx.?); + + @panic("No context attached to JSFSWatcher"); + } + + pub fn init(globalThis: *JSC.JSGlobalObject, manager: ?*FSWatcher.Watcher, fsevents_watcher: ?*FSEvents.FSEventsWatcher, signal: ?*JSC.AbortSignal, listener: JSC.JSValue, persistent: bool, encoding: JSC.Node.Encoding) !*JSObject { + var obj = try globalThis.allocator().create(JSObject); + obj.* = .{ + .signal = null, + .persistent = persistent, + .manager = manager, + .fsevents_watcher = fsevents_watcher, + .globalThis = globalThis, + .js_this = .zero, + .encoding = encoding, + .closed = false, + }; + const instance = obj.getFSWatcher(); + + if (persistent) { + obj.poll_ref.ref(instance.ctx); + } + + var js_this = JSObject.toJS(obj, globalThis); + JSObject.listenerSetCached(js_this, globalThis, listener); + obj.js_this = js_this; + obj.js_this.protect(); + + if (signal) |s| { + + // already aborted? + if (s.aborted()) { + obj.signal = s.ref(); + // abort next tick + var current_task: FSWatchTask = .{ + .ctx = instance, + }; + current_task.append("", .abort, .none); + current_task.enqueue(); + } else { + // watch for abortion + obj.signal = s.ref().listen(JSObject, obj, JSObject.emitAbort); + } + } + return obj; + } + + pub fn emitIfAborted(this: *JSObject) void { + if (this.signal) |s| { + if (s.aborted()) { + const err = s.abortReason(); + this.emitAbort(err); + } + } + } + + pub fn emitAbort(this: *JSObject, err: JSC.JSValue) void { + if (this.closed) return; + defer this.close(true); + + err.ensureStillAlive(); + + if (this.globalThis) |globalThis| { + if (this.js_this != .zero) { + if (JSObject.listenerGetCached(this.js_this)) |listener| { + var args = [_]JSC.JSValue{ + JSC.ZigString.static("error").toValue(globalThis), + if (err.isEmptyOrUndefinedOrNull()) JSC.WebCore.AbortSignal.createAbortError(JSC.ZigString.static("The user aborted a request"), &JSC.ZigString.Empty, globalThis) else err, + }; + _ = listener.callWithGlobalThis( + globalThis, + &args, + ); + } + } + } + } + pub fn emitError(this: *JSObject, err: string) void { + if (this.closed) return; + defer this.close(true); + + if (this.globalThis) |globalThis| { + if (this.js_this != .zero) { + if (JSObject.listenerGetCached(this.js_this)) |listener| { + var args = [_]JSC.JSValue{ + JSC.ZigString.static("error").toValue(globalThis), + JSC.ZigString.fromUTF8(err).toErrorInstance(globalThis), + }; + _ = listener.callWithGlobalThis( + globalThis, + &args, + ); + } + } + } + } + + pub fn emit(this: *JSObject, file_name: string, comptime eventType: string) void { + if (this.globalThis) |globalThis| { + if (this.js_this != .zero) { + if (JSObject.listenerGetCached(this.js_this)) |listener| { + var filename: JSC.JSValue = JSC.JSValue.jsUndefined(); + if (file_name.len > 0) { + if (this.encoding == .buffer) + filename = JSC.ArrayBuffer.createBuffer(globalThis, file_name) + else if (this.encoding == .utf8) { + filename = JSC.ZigString.fromUTF8(file_name).toValueGC(globalThis); + } else { + // convert to desired encoding + filename = Encoder.toStringAtRuntime(file_name.ptr, file_name.len, globalThis, this.encoding); + } + } + var args = [_]JSC.JSValue{ + JSC.ZigString.static(eventType).toValue(globalThis), + filename, + }; + _ = listener.callWithGlobalThis( + globalThis, + &args, + ); + } + } + } + } + + pub fn ref(this: *JSObject) void { + if (this.closed) return; + + if (!this.persistent) { + this.persistent = true; + this.poll_ref.ref(this.getFSWatcher().ctx); + } + } + + pub fn doRef(this: *JSObject, _: *JSC.JSGlobalObject, _: *JSC.CallFrame) callconv(.C) JSC.JSValue { + this.ref(); + return JSC.JSValue.jsUndefined(); + } + + pub fn unref(this: *JSObject) void { + if (this.persistent) { + this.persistent = false; + this.poll_ref.unref(this.getFSWatcher().ctx); + } + } + + pub fn doUnref(this: *JSObject, _: *JSC.JSGlobalObject, _: *JSC.CallFrame) callconv(.C) JSC.JSValue { + this.unref(); + return JSC.JSValue.jsUndefined(); + } + + pub fn hasRef(this: *JSObject, _: *JSC.JSGlobalObject, _: *JSC.CallFrame) callconv(.C) JSC.JSValue { + return JSC.JSValue.jsBoolean(this.persistent); + } + + pub fn close( + this: *JSObject, + emitEvent: bool, + ) void { + if (!this.closed) { + if (this.signal) |signal| { + this.signal = null; + signal.detach(this); + } + this.closed = true; + if (emitEvent) { + this.emit("", "close"); + } + + this.detach(); + } + } + + pub fn detach(this: *JSObject) void { + this.unref(); + + if (this.js_this != .zero) { + this.js_this.unprotect(); + this.js_this = .zero; + } + + this.globalThis = null; + + if (this.signal) |signal| { + this.signal = null; + signal.detach(this); + } + if (this.manager) |manager| { + var ctx = manager.ctx; + this.manager = null; + ctx.js_watcher = null; + ctx.deinit(); + manager.deinit(true); + } + + if (this.fsevents_watcher) |manager| { + var ctx = bun.cast(*FSWatcher, manager.ctx.?); + ctx.js_watcher = null; + ctx.deinit(); + manager.deinit(); + } + } + + pub fn doClose(this: *JSObject, _: *JSC.JSGlobalObject, _: *JSC.CallFrame) callconv(.C) JSC.JSValue { + this.close(true); + return JSC.JSValue.jsUndefined(); + } + + pub fn finalize(this: *JSObject) callconv(.C) void { + if (!this.closed) { + this.detach(); + } + + bun.default_allocator.destroy(this); + } + }; + + const PathResult = struct { + fd: StoredFileDescriptorType = 0, + is_file: bool = true, + }; + + fn fdFromAbsolutePathZ( + absolute_path_z: [:0]const u8, + ) !PathResult { + var stat = try bun.C.lstat_absolute(absolute_path_z); + var result = PathResult{}; + + switch (stat.kind) { + .sym_link => { + var file = try std.fs.openFileAbsoluteZ(absolute_path_z, .{ .mode = .read_only }); + result.fd = file.handle; + const _stat = try file.stat(); + + result.is_file = _stat.kind == .directory; + }, + .directory => { + const dir = (try std.fs.openIterableDirAbsoluteZ(absolute_path_z, .{ + .access_sub_paths = true, + })).dir; + result.fd = dir.fd; + result.is_file = false; + }, + else => { + const file = try std.fs.openFileAbsoluteZ(absolute_path_z, .{ .mode = .read_only }); + result.fd = file.handle; + result.is_file = true; + }, + } + return result; + } + + pub fn init(args: Arguments) !*FSWatcher { + var buf: [bun.MAX_PATH_BYTES + 1]u8 = undefined; + var slice = args.path.slice(); + if (bun.strings.startsWith(slice, "file://")) { + slice = slice[6..]; + } + var parts = [_]string{ + slice, + }; + + var file_path = Path.joinAbsStringBuf( + Fs.FileSystem.instance.top_level_dir, + &buf, + &parts, + .auto, + ); + + buf[file_path.len] = 0; + var file_path_z = buf[0..file_path.len :0]; + + var fs_type = try fdFromAbsolutePathZ(file_path_z); + + var ctx = try bun.default_allocator.create(FSWatcher); + const vm = args.global_this.bunVM(); + ctx.* = .{ + .ctx = vm, + .verbose = args.verbose, + .file_paths = bun.BabyList(string).initCapacity(bun.default_allocator, 1) catch |err| { + ctx.deinit(); + return err; + }, + }; + + if (comptime Environment.isMac) { + if (!fs_type.is_file) { + var dir_path_clone = bun.default_allocator.dupeZ(u8, file_path) catch unreachable; + ctx.entry_path = dir_path_clone; + ctx.entry_dir = dir_path_clone; + + var fsevents_watcher = FSEvents.watch(dir_path_clone, args.recursive, onFSEventUpdate, bun.cast(*anyopaque, ctx)) catch |err| { + ctx.deinit(); + return err; + }; + + ctx.js_watcher = JSObject.init(args.global_this, null, fsevents_watcher, args.signal, args.listener, args.persistent, args.encoding) catch |err| { + ctx.deinit(); + fsevents_watcher.deinit(); + return err; + }; + + return ctx; + } + } + + var fs_watcher = FSWatcher.Watcher.init( + ctx, + vm.bundler.fs, + bun.default_allocator, + ) catch |err| { + ctx.deinit(); + return err; + }; + + ctx.watcher_instance = fs_watcher; + + if (fs_type.is_file) { + var file_path_clone = bun.default_allocator.dupeZ(u8, file_path) catch unreachable; + + ctx.entry_path = file_path_clone; + ctx.entry_dir = std.fs.path.dirname(file_path_clone) orelse file_path_clone; + + fs_watcher.addFile(fs_type.fd, file_path_clone, FSWatcher.Watcher.getHash(file_path), options.Loader.file, 0, null, false) catch |err| { + ctx.deinit(); + fs_watcher.deinit(true); + return err; + }; + } else { + addDirectory(ctx, fs_watcher, fs_type.fd, file_path, args.recursive, &buf, true) catch |err| { + ctx.deinit(); + fs_watcher.deinit(true); + return err; + }; + } + + fs_watcher.start() catch |err| { + ctx.deinit(); + + fs_watcher.deinit(true); + return err; + }; + + ctx.js_watcher = JSObject.init(args.global_this, fs_watcher, null, args.signal, args.listener, args.persistent, args.encoding) catch |err| { + ctx.deinit(); + fs_watcher.deinit(true); + return err; + }; + + return ctx; + } +}; -- cgit v1.2.3 From 3ed28f2828a29129a1791b7a4f6935d842d6493c Mon Sep 17 00:00:00 2001 From: Ciro Spaciari Date: Sun, 25 Jun 2023 20:16:25 -0300 Subject: [fs.watch] fix reference/deinit (#3396) * fix js reference * fix close oops * refactor + hasPendingActivity * fmt * fix race conditions * fixup * add test calling close on error event * fix close inside close + test * cleanup --- src/bun.js/WebKit | 2 +- src/bun.js/bindings/JSSink.cpp | 2 +- src/bun.js/bindings/JSSink.h | 2 +- src/bun.js/bindings/ZigGeneratedClasses.cpp | 10 + src/bun.js/bindings/ZigGeneratedClasses.h | 27 ++ src/bun.js/bindings/generated_classes.zig | 1 + src/bun.js/bindings/generated_classes_list.zig | 2 +- src/bun.js/node/node.classes.ts | 1 + src/bun.js/node/node_fs_watcher.zig | 500 +++++++++++++------------ src/watcher.zig | 20 +- test/js/node/watch/fs.watch.test.js | 45 ++- 11 files changed, 352 insertions(+), 260 deletions(-) (limited to 'src/bun.js/node/node_fs_watcher.zig') diff --git a/src/bun.js/WebKit b/src/bun.js/WebKit index 4c8ab8fdf..b2f1006a0 160000 --- a/src/bun.js/WebKit +++ b/src/bun.js/WebKit @@ -1 +1 @@ -Subproject commit 4c8ab8fdfb102522fdd8e55d4eea53e8ce2755c2 +Subproject commit b2f1006a06f81bc860c89dd4c7cec3e7117c4c4c diff --git a/src/bun.js/bindings/JSSink.cpp b/src/bun.js/bindings/JSSink.cpp index 4acf01ff7..19bf05599 100644 --- a/src/bun.js/bindings/JSSink.cpp +++ b/src/bun.js/bindings/JSSink.cpp @@ -1,6 +1,6 @@ // AUTO-GENERATED FILE. DO NOT EDIT. -// Generated by 'make generate-sink' at 2023-06-14T21:38:04.394Z +// Generated by 'make generate-sink' at 2023-06-25T17:34:54.187Z // To regenerate this file, run: // // make generate-sink diff --git a/src/bun.js/bindings/JSSink.h b/src/bun.js/bindings/JSSink.h index 37c458e9b..9bf5554c4 100644 --- a/src/bun.js/bindings/JSSink.h +++ b/src/bun.js/bindings/JSSink.h @@ -1,6 +1,6 @@ // AUTO-GENERATED FILE. DO NOT EDIT. -// Generated by 'make generate-sink' at 2023-06-14T21:38:04.394Z +// Generated by 'make generate-sink' at 2023-06-25T17:34:54.186Z // #pragma once diff --git a/src/bun.js/bindings/ZigGeneratedClasses.cpp b/src/bun.js/bindings/ZigGeneratedClasses.cpp index e0a3f33d6..387580d54 100644 --- a/src/bun.js/bindings/ZigGeneratedClasses.cpp +++ b/src/bun.js/bindings/ZigGeneratedClasses.cpp @@ -5565,6 +5565,12 @@ void JSFSWatcherPrototype::finishCreation(JSC::VM& vm, JSC::JSGlobalObject* glob JSC_TO_STRING_TAG_WITHOUT_TRANSITION(); } +extern "C" bool FSWatcher__hasPendingActivity(void* ptr); +bool JSFSWatcher::hasPendingActivity(void* ctx) +{ + return FSWatcher__hasPendingActivity(ctx); +} + JSFSWatcher::~JSFSWatcher() { if (m_ctx) { @@ -5649,6 +5655,8 @@ void JSFSWatcher::visitChildrenImpl(JSCell* cell, Visitor& visitor) ASSERT_GC_OBJECT_INHERITS(thisObject, info()); Base::visitChildren(thisObject, visitor); visitor.append(thisObject->m_listener); + + visitor.addOpaqueRoot(thisObject->wrapped()); } DEFINE_VISIT_CHILDREN(JSFSWatcher); @@ -5659,6 +5667,8 @@ void JSFSWatcher::visitAdditionalChildren(Visitor& visitor) JSFSWatcher* thisObject = this; ASSERT_GC_OBJECT_INHERITS(thisObject, info()); visitor.append(thisObject->m_listener); + + visitor.addOpaqueRoot(this->wrapped()); } DEFINE_VISIT_ADDITIONAL_CHILDREN(JSFSWatcher); diff --git a/src/bun.js/bindings/ZigGeneratedClasses.h b/src/bun.js/bindings/ZigGeneratedClasses.h index 3fa0e26d2..1631f960e 100644 --- a/src/bun.js/bindings/ZigGeneratedClasses.h +++ b/src/bun.js/bindings/ZigGeneratedClasses.h @@ -623,10 +623,37 @@ public: : Base(vm, structure) { m_ctx = sinkPtr; + m_weakThis = JSC::Weak(this, getOwner()); } void finishCreation(JSC::VM&); + JSC::Weak m_weakThis; + + static bool hasPendingActivity(void* ctx); + + class Owner final : public JSC::WeakHandleOwner { + public: + bool isReachableFromOpaqueRoots(JSC::Handle handle, void* context, JSC::AbstractSlotVisitor& visitor, const char** reason) final + { + auto* controller = JSC::jsCast(handle.slot()->asCell()); + if (JSFSWatcher::hasPendingActivity(controller->wrapped())) { + if (UNLIKELY(reason)) + *reason = "has pending activity"; + return true; + } + + return visitor.containsOpaqueRoot(context); + } + void finalize(JSC::Handle, void* context) final {} + }; + + static JSC::WeakHandleOwner* getOwner() + { + static NeverDestroyed m_owner; + return &m_owner.get(); + } + DECLARE_VISIT_CHILDREN; template void visitAdditionalChildren(Visitor&); DECLARE_VISIT_OUTPUT_CONSTRAINTS; diff --git a/src/bun.js/bindings/generated_classes.zig b/src/bun.js/bindings/generated_classes.zig index 74e30cd83..bdde69c1a 100644 --- a/src/bun.js/bindings/generated_classes.zig +++ b/src/bun.js/bindings/generated_classes.zig @@ -1492,6 +1492,7 @@ pub const JSFSWatcher = struct { @export(FSWatcher.doRef, .{ .name = "FSWatcherPrototype__doRef" }); @export(FSWatcher.doUnref, .{ .name = "FSWatcherPrototype__doUnref" }); @export(FSWatcher.finalize, .{ .name = "FSWatcherClass__finalize" }); + @export(FSWatcher.hasPendingActivity, .{ .name = "FSWatcher__hasPendingActivity" }); @export(FSWatcher.hasRef, .{ .name = "FSWatcherPrototype__hasRef" }); } } diff --git a/src/bun.js/bindings/generated_classes_list.zig b/src/bun.js/bindings/generated_classes_list.zig index d90267337..543d492b5 100644 --- a/src/bun.js/bindings/generated_classes_list.zig +++ b/src/bun.js/bindings/generated_classes_list.zig @@ -37,5 +37,5 @@ pub const Classes = struct { pub const BuildArtifact = JSC.API.BuildArtifact; pub const BuildMessage = JSC.BuildMessage; pub const ResolveMessage = JSC.ResolveMessage; - pub const FSWatcher = JSC.Node.FSWatcher.JSObject; + pub const FSWatcher = JSC.Node.FSWatcher; }; diff --git a/src/bun.js/node/node.classes.ts b/src/bun.js/node/node.classes.ts index ce35c940a..2efad5245 100644 --- a/src/bun.js/node/node.classes.ts +++ b/src/bun.js/node/node.classes.ts @@ -7,6 +7,7 @@ export default [ noConstructor: true, finalize: true, configurable: false, + hasPendingActivity: true, klass: {}, JSType: "0b11101110", proto: { diff --git a/src/bun.js/node/node_fs_watcher.zig b/src/bun.js/node/node_fs_watcher.zig index 397d51916..b1f4ec8a9 100644 --- a/src/bun.js/node/node_fs_watcher.zig +++ b/src/bun.js/node/node_fs_watcher.zig @@ -4,6 +4,7 @@ const bun = @import("root").bun; const Fs = @import("../../fs.zig"); const Path = @import("../../resolver/resolve_path.zig"); const Encoder = JSC.WebCore.Encoder; +const Mutex = @import("../../lock.zig").Lock; const FSEvents = @import("./fs_events.zig"); @@ -30,17 +31,28 @@ pub const FSWatcher = struct { onAccept: std.ArrayHashMapUnmanaged(FSWatcher.Watcher.HashType, bun.BabyList(OnAcceptCallback), bun.ArrayIdentityContext, false) = .{}, ctx: *VirtualMachine, - js_watcher: ?*JSObject = null, - watcher_instance: ?*FSWatcher.Watcher = null, verbose: bool = false, file_paths: bun.BabyList(string) = .{}, entry_path: ?string = null, entry_dir: string = "", last_change_event: ChangeEvent = .{}, - pub fn toJS(this: *FSWatcher) JSC.JSValue { - return if (this.js_watcher) |js| js.js_this else JSC.JSValue.jsUndefined(); - } + // JSObject + mutex: Mutex, + signal: ?*JSC.AbortSignal, + persistent: bool, + default_watcher: ?*FSWatcher.Watcher, + fsevents_watcher: ?*FSEvents.FSEventsWatcher, + poll_ref: JSC.PollRef = .{}, + globalThis: *JSC.JSGlobalObject, + js_this: JSC.JSValue, + encoding: JSC.Node.Encoding, + // user can call close and pre-detach so we need to track this + closed: bool, + // counts pending tasks so we only deinit after all tasks are done + task_count: u32, + has_pending_activity: std.atomic.Atomic(bool), + pub usingnamespace JSC.Codegen.JSFSWatcher; pub fn eventLoop(this: FSWatcher) *EventLoop { return this.ctx.eventLoop(); @@ -51,6 +63,9 @@ pub const FSWatcher = struct { } pub fn deinit(this: *FSWatcher) void { + // stop all managers and signals + this.detach(); + while (this.file_paths.popOrNull()) |file_path| { bun.default_allocator.destroy(file_path); } @@ -107,41 +122,47 @@ pub const FSWatcher = struct { } pub fn run(this: *FSWatchTask) void { - // this runs on JS Context - if (this.ctx.js_watcher) |js_watcher| { - for (this.entries[0..this.count]) |entry| { - switch (entry.event_type) { - .rename => { - js_watcher.emit(entry.file_path, "rename"); - }, - .change => { - js_watcher.emit(entry.file_path, "change"); - }, - .@"error" => { - // file_path is the error message in this case - js_watcher.emitError(entry.file_path); - }, - .abort => { - js_watcher.emitIfAborted(); - }, - } + // this runs on JS Context Thread + + for (this.entries[0..this.count]) |entry| { + switch (entry.event_type) { + .rename => { + this.ctx.emit(entry.file_path, "rename"); + }, + .change => { + this.ctx.emit(entry.file_path, "change"); + }, + .@"error" => { + // file_path is the error message in this case + this.ctx.emitError(entry.file_path); + }, + .abort => { + this.ctx.emitIfAborted(); + }, } } + + this.ctx.unrefTask(); } pub fn enqueue(this: *FSWatchTask) void { if (this.count == 0) return; - var that = bun.default_allocator.create(FSWatchTask) catch unreachable; + // if false is closed or detached (can still contain valid refs but will not create a new one) + if (this.ctx.refTask()) { + var that = bun.default_allocator.create(FSWatchTask) catch unreachable; - that.* = this.*; - this.count = 0; - that.concurrent_task.task = JSC.Task.init(that); - this.ctx.enqueueTaskConcurrent(&that.concurrent_task); + that.* = this.*; + this.count = 0; + that.concurrent_task.task = JSC.Task.init(that); + this.ctx.enqueueTaskConcurrent(&that.concurrent_task); + return; + } + // closed or detached so just cleanEntries + this.cleanEntries(); } - - pub fn deinit(this: *FSWatchTask) void { + pub fn cleanEntries(this: *FSWatchTask) void { while (this.count > 0) { this.count -= 1; switch (this.entries[this.count].free_type) { @@ -150,6 +171,10 @@ pub const FSWatcher = struct { else => {}, } } + } + + pub fn deinit(this: *FSWatchTask) void { + this.cleanEntries(); bun.default_allocator.destroy(this); } }; @@ -275,7 +300,7 @@ pub const FSWatcher = struct { const kinds = slice.items(.kind); var _on_file_update_path_buf: [bun.MAX_PATH_BYTES]u8 = undefined; - var ctx = this.watcher_instance.?; + var ctx = this.default_watcher.?; defer ctx.flushEvictions(); defer Output.flush(); @@ -540,241 +565,225 @@ pub const FSWatcher = struct { pub fn createFSWatcher(this: Arguments) !JSC.JSValue { const obj = try FSWatcher.init(this); - return obj.toJS(); + if (obj.js_this != .zero) { + return obj.js_this; + } + return JSC.JSValue.jsUndefined(); } }; - pub const JSObject = struct { - signal: ?*JSC.AbortSignal, - persistent: bool, - manager: ?*FSWatcher.Watcher, - fsevents_watcher: ?*FSEvents.FSEventsWatcher, - poll_ref: JSC.PollRef = .{}, - globalThis: ?*JSC.JSGlobalObject, - js_this: JSC.JSValue, - encoding: JSC.Node.Encoding, - closed: bool, - - pub usingnamespace JSC.Codegen.JSFSWatcher; - - pub fn getFSWatcher(this: *JSObject) *FSWatcher { - if (this.manager) |manager| return manager.ctx; - if (this.fsevents_watcher) |manager| return bun.cast(*FSWatcher, manager.ctx.?); - - @panic("No context attached to JSFSWatcher"); + pub fn initJS(this: *FSWatcher, listener: JSC.JSValue) void { + if (this.persistent) { + this.poll_ref.ref(this.ctx); } - pub fn init(globalThis: *JSC.JSGlobalObject, manager: ?*FSWatcher.Watcher, fsevents_watcher: ?*FSEvents.FSEventsWatcher, signal: ?*JSC.AbortSignal, listener: JSC.JSValue, persistent: bool, encoding: JSC.Node.Encoding) !*JSObject { - var obj = try globalThis.allocator().create(JSObject); - obj.* = .{ - .signal = null, - .persistent = persistent, - .manager = manager, - .fsevents_watcher = fsevents_watcher, - .globalThis = globalThis, - .js_this = .zero, - .encoding = encoding, - .closed = false, - }; - const instance = obj.getFSWatcher(); - - if (persistent) { - obj.poll_ref.ref(instance.ctx); - } - - var js_this = JSObject.toJS(obj, globalThis); - JSObject.listenerSetCached(js_this, globalThis, listener); - obj.js_this = js_this; - obj.js_this.protect(); - - if (signal) |s| { - - // already aborted? - if (s.aborted()) { - obj.signal = s.ref(); - // abort next tick - var current_task: FSWatchTask = .{ - .ctx = instance, - }; - current_task.append("", .abort, .none); - current_task.enqueue(); - } else { - // watch for abortion - obj.signal = s.ref().listen(JSObject, obj, JSObject.emitAbort); - } + const js_this = FSWatcher.toJS(this, this.globalThis); + js_this.ensureStillAlive(); + this.js_this = js_this; + FSWatcher.listenerSetCached(js_this, this.globalThis, listener); + + if (this.signal) |s| { + // already aborted? + if (s.aborted()) { + // safely abort next tick + var current_task: FSWatchTask = .{ + .ctx = this, + }; + current_task.append("", .abort, .none); + current_task.enqueue(); + } else { + // watch for abortion + this.signal = s.listen(FSWatcher, this, FSWatcher.emitAbort); } - return obj; } + } - pub fn emitIfAborted(this: *JSObject) void { - if (this.signal) |s| { - if (s.aborted()) { - const err = s.abortReason(); - this.emitAbort(err); - } + pub fn emitIfAborted(this: *FSWatcher) void { + if (this.signal) |s| { + if (s.aborted()) { + const err = s.abortReason(); + this.emitAbort(err); } } + } - pub fn emitAbort(this: *JSObject, err: JSC.JSValue) void { - if (this.closed) return; - defer this.close(true); - - err.ensureStillAlive(); - - if (this.globalThis) |globalThis| { - if (this.js_this != .zero) { - if (JSObject.listenerGetCached(this.js_this)) |listener| { - var args = [_]JSC.JSValue{ - JSC.ZigString.static("error").toValue(globalThis), - if (err.isEmptyOrUndefinedOrNull()) JSC.WebCore.AbortSignal.createAbortError(JSC.ZigString.static("The user aborted a request"), &JSC.ZigString.Empty, globalThis) else err, - }; - _ = listener.callWithGlobalThis( - globalThis, - &args, - ); - } - } + pub fn emitAbort(this: *FSWatcher, err: JSC.JSValue) void { + if (this.closed) return; + defer this.close(); + + err.ensureStillAlive(); + if (this.js_this != .zero) { + const js_this = this.js_this; + js_this.ensureStillAlive(); + if (FSWatcher.listenerGetCached(js_this)) |listener| { + listener.ensureStillAlive(); + var args = [_]JSC.JSValue{ + JSC.ZigString.static("error").toValue(this.globalThis), + if (err.isEmptyOrUndefinedOrNull()) JSC.WebCore.AbortSignal.createAbortError(JSC.ZigString.static("The user aborted a request"), &JSC.ZigString.Empty, this.globalThis) else err, + }; + _ = listener.callWithGlobalThis( + this.globalThis, + &args, + ); } } - pub fn emitError(this: *JSObject, err: string) void { - if (this.closed) return; - defer this.close(true); - - if (this.globalThis) |globalThis| { - if (this.js_this != .zero) { - if (JSObject.listenerGetCached(this.js_this)) |listener| { - var args = [_]JSC.JSValue{ - JSC.ZigString.static("error").toValue(globalThis), - JSC.ZigString.fromUTF8(err).toErrorInstance(globalThis), - }; - _ = listener.callWithGlobalThis( - globalThis, - &args, - ); - } - } + } + pub fn emitError(this: *FSWatcher, err: string) void { + if (this.closed) return; + defer this.close(); + + if (this.js_this != .zero) { + const js_this = this.js_this; + js_this.ensureStillAlive(); + if (FSWatcher.listenerGetCached(js_this)) |listener| { + listener.ensureStillAlive(); + var args = [_]JSC.JSValue{ + JSC.ZigString.static("error").toValue(this.globalThis), + JSC.ZigString.fromUTF8(err).toErrorInstance(this.globalThis), + }; + _ = listener.callWithGlobalThis( + this.globalThis, + &args, + ); } } + } - pub fn emit(this: *JSObject, file_name: string, comptime eventType: string) void { - if (this.globalThis) |globalThis| { - if (this.js_this != .zero) { - if (JSObject.listenerGetCached(this.js_this)) |listener| { - var filename: JSC.JSValue = JSC.JSValue.jsUndefined(); - if (file_name.len > 0) { - if (this.encoding == .buffer) - filename = JSC.ArrayBuffer.createBuffer(globalThis, file_name) - else if (this.encoding == .utf8) { - filename = JSC.ZigString.fromUTF8(file_name).toValueGC(globalThis); - } else { - // convert to desired encoding - filename = Encoder.toStringAtRuntime(file_name.ptr, file_name.len, globalThis, this.encoding); - } - } - var args = [_]JSC.JSValue{ - JSC.ZigString.static(eventType).toValue(globalThis), - filename, - }; - _ = listener.callWithGlobalThis( - globalThis, - &args, - ); + pub fn emit(this: *FSWatcher, file_name: string, comptime eventType: string) void { + if (this.js_this != .zero) { + const js_this = this.js_this; + js_this.ensureStillAlive(); + if (FSWatcher.listenerGetCached(js_this)) |listener| { + listener.ensureStillAlive(); + var filename: JSC.JSValue = JSC.JSValue.jsUndefined(); + if (file_name.len > 0) { + if (this.encoding == .buffer) + filename = JSC.ArrayBuffer.createBuffer(this.globalThis, file_name) + else if (this.encoding == .utf8) { + filename = JSC.ZigString.fromUTF8(file_name).toValueGC(this.globalThis); + } else { + // convert to desired encoding + filename = Encoder.toStringAtRuntime(file_name.ptr, file_name.len, this.globalThis, this.encoding); } } + var args = [_]JSC.JSValue{ + JSC.ZigString.static(eventType).toValue(this.globalThis), + filename, + }; + _ = listener.callWithGlobalThis( + this.globalThis, + &args, + ); } } + } - pub fn ref(this: *JSObject) void { - if (this.closed) return; - - if (!this.persistent) { - this.persistent = true; - this.poll_ref.ref(this.getFSWatcher().ctx); - } - } - - pub fn doRef(this: *JSObject, _: *JSC.JSGlobalObject, _: *JSC.CallFrame) callconv(.C) JSC.JSValue { - this.ref(); - return JSC.JSValue.jsUndefined(); + pub fn doRef(this: *FSWatcher, _: *JSC.JSGlobalObject, _: *JSC.CallFrame) callconv(.C) JSC.JSValue { + if (!this.closed and !this.persistent) { + this.persistent = true; + this.poll_ref.ref(this.ctx); } + return JSC.JSValue.jsUndefined(); + } - pub fn unref(this: *JSObject) void { - if (this.persistent) { - this.persistent = false; - this.poll_ref.unref(this.getFSWatcher().ctx); - } + pub fn doUnref(this: *FSWatcher, _: *JSC.JSGlobalObject, _: *JSC.CallFrame) callconv(.C) JSC.JSValue { + if (this.persistent) { + this.persistent = false; + this.poll_ref.unref(this.ctx); } + return JSC.JSValue.jsUndefined(); + } - pub fn doUnref(this: *JSObject, _: *JSC.JSGlobalObject, _: *JSC.CallFrame) callconv(.C) JSC.JSValue { - this.unref(); - return JSC.JSValue.jsUndefined(); - } + pub fn hasRef(this: *FSWatcher, _: *JSC.JSGlobalObject, _: *JSC.CallFrame) callconv(.C) JSC.JSValue { + return JSC.JSValue.jsBoolean(this.persistent); + } - pub fn hasRef(this: *JSObject, _: *JSC.JSGlobalObject, _: *JSC.CallFrame) callconv(.C) JSC.JSValue { - return JSC.JSValue.jsBoolean(this.persistent); - } + // this can be called from Watcher Thread or JS Context Thread + pub fn refTask(this: *FSWatcher) bool { + this.mutex.lock(); + defer this.mutex.unlock(); + // stop new references + if (this.closed) return false; + this.task_count += 1; + return true; + } - pub fn close( - this: *JSObject, - emitEvent: bool, - ) void { - if (!this.closed) { - if (this.signal) |signal| { - this.signal = null; - signal.detach(this); - } - this.closed = true; - if (emitEvent) { - this.emit("", "close"); - } + pub fn hasPendingActivity(this: *FSWatcher) callconv(.C) bool { + @fence(.Acquire); + return this.has_pending_activity.load(.Acquire); + } + // only called from Main Thread + pub fn updateHasPendingActivity(this: *FSWatcher) void { + @fence(.Release); + this.has_pending_activity.store(false, .Release); + } - this.detach(); - } + // unref is always called on main JS Context Thread + pub fn unrefTask(this: *FSWatcher) void { + this.mutex.lock(); + defer this.mutex.unlock(); + this.task_count -= 1; + if (this.closed and this.task_count == 0) { + this.updateHasPendingActivity(); } + } - pub fn detach(this: *JSObject) void { - this.unref(); + pub fn close( + this: *FSWatcher, + ) void { + this.mutex.lock(); + if (!this.closed) { + this.closed = true; - if (this.js_this != .zero) { - this.js_this.unprotect(); - this.js_this = .zero; - } + // emit should only be called unlocked + this.mutex.unlock(); - this.globalThis = null; + this.emit("", "close"); + // we immediately detach here + this.detach(); - if (this.signal) |signal| { - this.signal = null; - signal.detach(this); - } - if (this.manager) |manager| { - var ctx = manager.ctx; - this.manager = null; - ctx.js_watcher = null; - ctx.deinit(); - manager.deinit(true); + // no need to lock again, because ref checks closed and unref is only called on main thread + if (this.task_count == 0) { + this.updateHasPendingActivity(); } + } else { + this.mutex.unlock(); + } + } - if (this.fsevents_watcher) |manager| { - var ctx = bun.cast(*FSWatcher, manager.ctx.?); - ctx.js_watcher = null; - ctx.deinit(); - manager.deinit(); - } + // this can be called multiple times + pub fn detach(this: *FSWatcher) void { + if (this.persistent) { + this.persistent = false; + this.poll_ref.unref(this.ctx); } - pub fn doClose(this: *JSObject, _: *JSC.JSGlobalObject, _: *JSC.CallFrame) callconv(.C) JSC.JSValue { - this.close(true); - return JSC.JSValue.jsUndefined(); + if (this.signal) |signal| { + this.signal = null; + signal.detach(this); } - pub fn finalize(this: *JSObject) callconv(.C) void { - if (!this.closed) { - this.detach(); - } + if (this.default_watcher) |default_watcher| { + this.default_watcher = null; + default_watcher.deinit(true); + } - bun.default_allocator.destroy(this); + if (this.fsevents_watcher) |fsevents_watcher| { + this.fsevents_watcher = null; + fsevents_watcher.deinit(); } - }; + + this.js_this = .zero; + } + + pub fn doClose(this: *FSWatcher, _: *JSC.JSGlobalObject, _: *JSC.CallFrame) callconv(.C) JSC.JSValue { + this.close(); + return JSC.JSValue.jsUndefined(); + } + + pub fn finalize(this: *FSWatcher) callconv(.C) void { + this.deinit(); + } const PathResult = struct { fd: StoredFileDescriptorType = 0, @@ -837,6 +846,17 @@ pub const FSWatcher = struct { const vm = args.global_this.bunVM(); ctx.* = .{ .ctx = vm, + .mutex = Mutex.init(), + .signal = if (args.signal) |s| s.ref() else null, + .persistent = args.persistent, + .default_watcher = null, + .fsevents_watcher = null, + .globalThis = args.global_this, + .js_this = .zero, + .encoding = args.encoding, + .closed = false, + .task_count = 0, + .has_pending_activity = std.atomic.Atomic(bool).init(true), .verbose = args.verbose, .file_paths = bun.BabyList(string).initCapacity(bun.default_allocator, 1) catch |err| { ctx.deinit(); @@ -850,22 +870,17 @@ pub const FSWatcher = struct { ctx.entry_path = dir_path_clone; ctx.entry_dir = dir_path_clone; - var fsevents_watcher = FSEvents.watch(dir_path_clone, args.recursive, onFSEventUpdate, bun.cast(*anyopaque, ctx)) catch |err| { + ctx.fsevents_watcher = FSEvents.watch(dir_path_clone, args.recursive, onFSEventUpdate, bun.cast(*anyopaque, ctx)) catch |err| { ctx.deinit(); return err; }; - ctx.js_watcher = JSObject.init(args.global_this, null, fsevents_watcher, args.signal, args.listener, args.persistent, args.encoding) catch |err| { - ctx.deinit(); - fsevents_watcher.deinit(); - return err; - }; - + ctx.initJS(args.listener); return ctx; } } - var fs_watcher = FSWatcher.Watcher.init( + var default_watcher = FSWatcher.Watcher.init( ctx, vm.bundler.fs, bun.default_allocator, @@ -874,7 +889,7 @@ pub const FSWatcher = struct { return err; }; - ctx.watcher_instance = fs_watcher; + ctx.default_watcher = default_watcher; if (fs_type.is_file) { var file_path_clone = bun.default_allocator.dupeZ(u8, file_path) catch unreachable; @@ -882,32 +897,23 @@ pub const FSWatcher = struct { ctx.entry_path = file_path_clone; ctx.entry_dir = std.fs.path.dirname(file_path_clone) orelse file_path_clone; - fs_watcher.addFile(fs_type.fd, file_path_clone, FSWatcher.Watcher.getHash(file_path), options.Loader.file, 0, null, false) catch |err| { + default_watcher.addFile(fs_type.fd, file_path_clone, FSWatcher.Watcher.getHash(file_path), options.Loader.file, 0, null, false) catch |err| { ctx.deinit(); - fs_watcher.deinit(true); return err; }; } else { - addDirectory(ctx, fs_watcher, fs_type.fd, file_path, args.recursive, &buf, true) catch |err| { + addDirectory(ctx, default_watcher, fs_type.fd, file_path, args.recursive, &buf, true) catch |err| { ctx.deinit(); - fs_watcher.deinit(true); return err; }; } - fs_watcher.start() catch |err| { + default_watcher.start() catch |err| { ctx.deinit(); - - fs_watcher.deinit(true); - return err; - }; - - ctx.js_watcher = JSObject.init(args.global_this, fs_watcher, null, args.signal, args.listener, args.persistent, args.encoding) catch |err| { - ctx.deinit(); - fs_watcher.deinit(true); return err; }; + ctx.initJS(args.listener); return ctx; } }; diff --git a/src/watcher.zig b/src/watcher.zig index 044770dc4..e3b3600ad 100644 --- a/src/watcher.zig +++ b/src/watcher.zig @@ -519,7 +519,7 @@ pub fn NewWatcher(comptime ContextType: type) type { var changelist_array: [128]KEvent = std.mem.zeroes([128]KEvent); var changelist = &changelist_array; - while (this.running) { + while (true) { defer Output.flush(); var count_ = std.os.system.kevent( @@ -576,10 +576,12 @@ pub fn NewWatcher(comptime ContextType: type) type { defer this.mutex.unlock(); if (this.running) { this.ctx.onFileUpdate(watchevents, this.changed_filepaths[0..watchevents.len], this.watchlist); + } else { + break; } } } else if (Environment.isLinux) { - restart: while (this.running) { + restart: while (true) { defer Output.flush(); var events = try INotify.read(); @@ -588,14 +590,14 @@ pub fn NewWatcher(comptime ContextType: type) type { // TODO: is this thread safe? var remaining_events = events.len; - var name_off: u8 = 0; - var temp_name_list: [128]?[:0]u8 = undefined; - var temp_name_off: u8 = 0; - const eventlist_index = this.watchlist.items(.eventlist_index); while (remaining_events > 0) { - const slice = events[0..@min(remaining_events, this.watch_events.len)]; + var name_off: u8 = 0; + var temp_name_list: [128]?[:0]u8 = undefined; + var temp_name_off: u8 = 0; + + const slice = events[0..@min(128, remaining_events, this.watch_events.len)]; var watchevents = this.watch_events[0..slice.len]; var watch_event_id: u32 = 0; for (slice) |event| { @@ -647,8 +649,10 @@ pub fn NewWatcher(comptime ContextType: type) type { defer this.mutex.unlock(); if (this.running) { this.ctx.onFileUpdate(all_events[0 .. last_event_index + 1], this.changed_filepaths[0 .. name_off + 1], this.watchlist); - remaining_events -= slice.len; + } else { + break; } + remaining_events -= slice.len; } } } diff --git a/test/js/node/watch/fs.watch.test.js b/test/js/node/watch/fs.watch.test.js index 56e1798f1..33d05df29 100644 --- a/test/js/node/watch/fs.watch.test.js +++ b/test/js/node/watch/fs.watch.test.js @@ -17,6 +17,8 @@ const testDir = tempDirWithFiles("watch", { "relative.txt": "hello", "abort.txt": "hello", "url.txt": "hello", + "close.txt": "hello", + "close-close.txt": "hello", [encodingFileName]: "hello", }); @@ -105,6 +107,7 @@ describe("fs.watch", () => { let err = undefined; watcher.on("change", (event, filename) => { const basename = path.basename(filename); + if (basename === "subfolder") return; count++; try { @@ -274,6 +277,46 @@ describe("fs.watch", () => { } }); + test("calling close from error event should not throw", done => { + const filepath = path.join(testDir, "close.txt"); + try { + const ac = new AbortController(); + const watcher = fs.watch(pathToFileURL(filepath), { signal: ac.signal }); + watcher.once("error", () => { + try { + watcher.close(); + done(); + } catch (e) { + done("Should not error when calling close from error event"); + } + }); + ac.abort(); + } catch (e) { + done(e); + } + }); + + test("calling close from close event should not throw", done => { + const filepath = path.join(testDir, "close-close.txt"); + try { + const ac = new AbortController(); + const watcher = fs.watch(pathToFileURL(filepath), { signal: ac.signal }); + + watcher.once("close", () => { + try { + watcher.close(); + done(); + } catch (e) { + done("Should not error when calling close from close event"); + } + }); + + ac.abort(); + } catch (e) { + done(e); + } + }); + test("Signal aborted after creating the watcher", async () => { const filepath = path.join(testDir, "abort.txt"); @@ -300,7 +343,7 @@ describe("fs.watch", () => { }); }); -describe("fs.promises.watchFile", () => { +describe("fs.promises.watch", () => { test("add file/folder to folder", async () => { let count = 0; const root = path.join(testDir, "add-promise-directory"); -- cgit v1.2.3 From 034577c9dad2486324e3bd65907794dd9a3ee217 Mon Sep 17 00:00:00 2001 From: Ciro Spaciari Date: Mon, 3 Jul 2023 18:08:49 -0300 Subject: [fix] patch fs.watch sym link (#3481) * oopsie * add tests --------- Co-authored-by: Jarred Sumner --- src/bun.js/node/node_fs_watcher.zig | 2 +- test/js/node/watch/fs.watch.test.js | 55 +++++++++++++++++++++++++++++++++++++ 2 files changed, 56 insertions(+), 1 deletion(-) (limited to 'src/bun.js/node/node_fs_watcher.zig') diff --git a/src/bun.js/node/node_fs_watcher.zig b/src/bun.js/node/node_fs_watcher.zig index b1f4ec8a9..d0af350c0 100644 --- a/src/bun.js/node/node_fs_watcher.zig +++ b/src/bun.js/node/node_fs_watcher.zig @@ -802,7 +802,7 @@ pub const FSWatcher = struct { result.fd = file.handle; const _stat = try file.stat(); - result.is_file = _stat.kind == .directory; + result.is_file = _stat.kind != .directory; }, .directory => { const dir = (try std.fs.openIterableDirAbsoluteZ(absolute_path_z, .{ diff --git a/test/js/node/watch/fs.watch.test.js b/test/js/node/watch/fs.watch.test.js index 33d05df29..faf6a8546 100644 --- a/test/js/node/watch/fs.watch.test.js +++ b/test/js/node/watch/fs.watch.test.js @@ -19,6 +19,8 @@ const testDir = tempDirWithFiles("watch", { "url.txt": "hello", "close.txt": "hello", "close-close.txt": "hello", + "sym-sync.txt": "hello", + "sym.txt": "hello", [encodingFileName]: "hello", }); @@ -341,6 +343,36 @@ describe("fs.watch", () => { watcher.once("close", () => reject()); }); }); + + test("should work with symlink", async () => { + const filepath = path.join(testDir, "sym-symlink2.txt"); + await fs.promises.symlink(path.join(testDir, "sym-sync.txt"), filepath); + + const interval = repeat(() => { + fs.writeFileSync(filepath, "hello"); + }); + + const promise = new Promise((resolve, reject) => { + let timeout = null; + const watcher = fs.watch(filepath, event => { + clearTimeout(timeout); + clearInterval(interval); + try { + resolve(event); + } catch (e) { + reject(e); + } finally { + watcher.close(); + } + }); + setTimeout(() => { + clearInterval(interval); + watcher?.close(); + reject("timeout"); + }, 3000); + }); + expect(promise).resolves.toBe("change"); + }); }); describe("fs.promises.watch", () => { @@ -464,4 +496,27 @@ describe("fs.promises.watch", () => { } })(); }); + + test("should work with symlink", async () => { + const filepath = path.join(testDir, "sym-symlink.txt"); + await fs.promises.symlink(path.join(testDir, "sym.txt"), filepath); + + const watcher = fs.promises.watch(filepath); + const interval = repeat(() => { + fs.writeFileSync(filepath, "hello"); + }); + + const promise = (async () => { + try { + for await (const event of watcher) { + return event.eventType; + } + } catch (e) { + expect("unreacheable").toBe(false); + } finally { + clearInterval(interval); + } + })(); + expect(promise).resolves.toBe("change"); + }); }); -- cgit v1.2.3