Commit 73b17474d7ff620fa68434e233b472269bdca2b4 | zig

author	Andrew Kelley <andrew@ziglang.org> 2021-01-11 12:45:36-08:00
committer	GitHub <noreply@github.com> 2021-01-11 12:45:36-08:00
log	73b17474d7ff620fa68434e233b472269bdca2b4
tree	8b5a1589b891a943572e3a98c8504ce2f9ddc6be
parent	cc2981edfc4c0e36d3ce6d564f7a36caae9b61b7
parent	16144a7a37c2b2059d8a2fcfc70d6cf0ab49ab00
signature	Signed by PGP key 4AEE18F83AFDEB23

Merge pull request #7134 from alexnask/fix_std_fs_watch

The std.fs.Watch rewrite PR

5 files changed, 402 insertions(+), 356 deletions(-)

lib
- std
  - fs
    watch.zig
  - os
    bits
    freebsd.zig
    windows.zig
    windows
    bits.zig
    kernel32.zig

lib/std/fs/watch.zig+390-349

...	...	@@ -3,7 +3,7 @@
3	3	// This file is part of [zig](https://ziglang.org/), which is MIT licensed.
4	4	// The MIT license requires this copyright notice to be included in all copies
5	5	// and substantial portions of the software.
6		const std = @import("../std.zig");
	6	const std = @import("std");
7	7	const builtin = @import("builtin");
8	8	const event = std.event;
9	9	const assert = std.debug.assert;
...	...	@@ -24,16 +24,6 @@ const WatchEventId = enum {
24	24	Delete,
25	25	};
26	26
27		fn eqlString(a: []const u16, b: []const u16) bool {
28		if (a.len != b.len) return false;
29		if (a.ptr == b.ptr) return true;
30		return mem.compare(u16, a, b) == .Equal;
31		}
32
33		fn hashString(s: []const u16) u32 {
34		return @truncate(u32, std.hash.Wyhash.hash(0, mem.sliceAsBytes(s)));
35		}
36
37	27	const WatchEventError = error{
38	28	UserResourceLimitReached,
39	29	SystemResources,
...	...	@@ -43,7 +33,7 @@ const WatchEventError = error{
43	33
44	34	pub fn Watch(comptime V: type) type {
45	35	return struct {
46		channel: *event.Channel(Event.Error!Event),
	36	channel: event.Channel(Event.Error!Event),
47	37	os_data: OsData,
48	38	allocator: *Allocator,
49	39
...	...	@@ -57,10 +47,10 @@ pub fn Watch(comptime V: type) type {
57	47	};
58	48
59	49	const KqOsData = struct {
60		file_table: FileTable,
61	50	table_lock: event.Lock,
	51	file_table: FileTable,
62	52
63		const FileTable = std.StringHashMap(*Put);
	53	const FileTable = std.StringHashMapUnmanaged(*Put);
64	54	const Put = struct {
65	55	putter_frame: @Frame(kqPutEvents),
66	56	cancelled: bool = false,
...	...	@@ -71,21 +61,15 @@ pub fn Watch(comptime V: type) type {
71	61	const WindowsOsData = struct {
72	62	table_lock: event.Lock,
73	63	dir_table: DirTable,
74		all_putters: std.atomic.Queue(Put),
75		ref_count: std.atomic.Int(usize),
76
77		const Put = struct {
78		putter: anyframe,
79		cancelled: bool = false,
80		};
	64	cancelled: bool = false,
81	65
82		const DirTable = std.StringHashMap(*Dir);
83		const FileTable = std.HashMap([]const u16, V, hashString, eqlString);
	66	const DirTable = std.StringHashMapUnmanaged(*Dir);
	67	const FileTable = std.StringHashMapUnmanaged(V);
84	68
85	69	const Dir = struct {
86	70	putter_frame: @Frame(windowsDirReader),
87	71	file_table: FileTable,
88		table_lock: event.Lock,
	72	dir_handle: os.windows.HANDLE,
89	73	};
90	74	};
91	75
...	...	@@ -96,8 +80,8 @@ pub fn Watch(comptime V: type) type {
96	80	table_lock: event.Lock,
97	81	cancelled: bool = false,
98	82
99		const WdTable = std.AutoHashMap(i32, Dir);
100		const FileTable = std.StringHashMap(V);
	83	const WdTable = std.AutoHashMapUnmanaged(i32, Dir);
	84	const FileTable = std.StringHashMapUnmanaged(V);
101	85
102	86	const Dir = struct {
103	87	dirname: []const u8,
...	...	@@ -110,19 +94,14 @@ pub fn Watch(comptime V: type) type {
110	94	pub const Event = struct {
111	95	id: Id,
112	96	data: V,
	97	dirname: []const u8,
	98	basename: []const u8,
113	99
114	100	pub const Id = WatchEventId;
115	101	pub const Error = WatchEventError;
116	102	};
117	103
118	104	pub fn init(allocator: Allocator, event_buf_count: usize) !Self {
119		const channel = try allocator.create(event.Channel(Event.Error!Event));
120		errdefer allocator.destroy(channel);
121		var buf = try allocator.alloc(Event.Error!Event, event_buf_count);
122		errdefer allocator.free(buf);
123		channel.init(buf);
124		errdefer channel.deinit();
125
126	105	const self = try allocator.create(Self);
127	106	errdefer allocator.destroy(self);
128	107
...	...	@@ -133,15 +112,17 @@ pub fn Watch(comptime V: type) type {
133	112
134	113	self.* = Self{
135	114	.allocator = allocator,
136		.channel = channel,
	115	.channel = undefined,
137	116	.os_data = OsData{
138	117	.putter_frame = undefined,
139	118	.inotify_fd = inotify_fd,
140	119	.wd_table = OsData.WdTable.init(allocator),
141		.table_lock = event.Lock.init(),
	120	.table_lock = event.Lock{},
142	121	},
143	122	};
144	123
	124	var buf = try allocator.alloc(Event.Error!Event, event_buf_count);
	125	self.channel.init(buf);
145	126	self.os_data.putter_frame = async self.linuxEventPutter();
146	127	return self;
147	128	},
...	...	@@ -149,82 +130,93 @@ pub fn Watch(comptime V: type) type {
149	130	.windows => {
150	131	self.* = Self{
151	132	.allocator = allocator,
152		.channel = channel,
	133	.channel = undefined,
153	134	.os_data = OsData{
154		.table_lock = event.Lock.init(),
	135	.table_lock = event.Lock{},
155	136	.dir_table = OsData.DirTable.init(allocator),
156		.ref_count = std.atomic.Int(usize).init(1),
157		.all_putters = std.atomic.Queue(anyframe).init(),
158	137	},
159	138	};
	139
	140	var buf = try allocator.alloc(Event.Error!Event, event_buf_count);
	141	self.channel.init(buf);
160	142	return self;
161	143	},
162	144
163	145	.macos, .freebsd, .netbsd, .dragonfly, .openbsd => {
164	146	self.* = Self{
165	147	.allocator = allocator,
166		.channel = channel,
	148	.channel = undefined,
167	149	.os_data = OsData{
168		.table_lock = event.Lock.init(),
	150	.table_lock = event.Lock{},
169	151	.file_table = OsData.FileTable.init(allocator),
170	152	},
171	153	};
	154
	155	var buf = try allocator.alloc(Event.Error!Event, event_buf_count);
	156	self.channel.init(buf);
172	157	return self;
173	158	},
174	159	else => @compileError("Unsupported OS"),
175	160	}
176	161	}
177	162
178		/// All addFile calls and removeFile calls must have completed.
179	163	pub fn deinit(self: *Self) void {
180	164	switch (builtin.os.tag) {
181	165	.macos, .freebsd, .netbsd, .dragonfly, .openbsd => {
182		// TODO we need to cancel the frames before destroying the lock
183		self.os_data.table_lock.deinit();
184	166	var it = self.os_data.file_table.iterator();
185	167	while (it.next()) \|entry\| {
186		entry.cancelled = true;
187		await entry.value.putter;
	168	entry.value.cancelled = true;
	169	// @TODO Close the fd here?
	170	await entry.value.putter_frame;
188	171	self.allocator.free(entry.key);
189		self.allocator.free(entry.value);
	172	self.allocator.destroy(entry.value);
190	173	}
191		self.channel.deinit();
192		self.allocator.destroy(self.channel.buffer_nodes);
193		self.allocator.destroy(self);
194	174	},
195	175	.linux => {
196	176	self.os_data.cancelled = true;
	177	{
	178	// Remove all directory watches linuxEventPutter will take care of
	179	// cleaning up the memory and closing the inotify fd.
	180	var dir_it = self.os_data.wd_table.iterator();
	181	while (dir_it.next()) \|wd_entry\| {
	182	const rc = os.linux.inotify_rm_watch(self.os_data.inotify_fd, wd_entry.key);
	183	// Errno can only be EBADF, EINVAL if either the inotify fs or the wd are invalid
	184	std.debug.assert(rc == 0);
	185	}
	186	}
197	187	await self.os_data.putter_frame;
198		self.allocator.destroy(self);
199	188	},
200	189	.windows => {
201		while (self.os_data.all_putters.get()) \|putter_node\| {
202		putter_node.cancelled = true;
203		await putter_node.frame;
	190	self.os_data.cancelled = true;
	191	var dir_it = self.os_data.dir_table.iterator();
	192	while (dir_it.next()) \|dir_entry\| {
	193	if (windows.kernel32.CancelIoEx(dir_entry.value.dir_handle, null) != 0) {
	194	// We canceled the pending ReadDirectoryChangesW operation, but our
	195	// frame is still suspending, now waiting indefinitely.
	196	// Thus, it is safe to resume it ourslves
	197	resume dir_entry.value.putter_frame;
	198	} else {
	199	std.debug.assert(windows.kernel32.GetLastError() == .NOT_FOUND);
	200	// We are at another suspend point, we can await safely for the
	201	// function to exit the loop
	202	await dir_entry.value.putter_frame;
	203	}
	204
	205	self.allocator.free(dir_entry.key);
	206	var file_it = dir_entry.value.file_table.iterator();
	207	while (file_it.next()) \|file_entry\| {
	208	self.allocator.free(file_entry.key);
	209	}
	210	dir_entry.value.file_table.deinit(self.allocator);
	211	self.allocator.destroy(dir_entry.value);
204	212	}
205		self.deref();
	213	self.os_data.dir_table.deinit(self.allocator);
206	214	},
207	215	else => @compileError("Unsupported OS"),
208	216	}
209		}
210
211		fn ref(self: *Self) void {
212		_ = self.os_data.ref_count.incr();
213		}
214
215		fn deref(self: *Self) void {
216		if (self.os_data.ref_count.decr() == 1) {
217		self.os_data.table_lock.deinit();
218		var it = self.os_data.dir_table.iterator();
219		while (it.next()) \|entry\| {
220		self.allocator.free(entry.key);
221		self.allocator.destroy(entry.value);
222		}
223		self.os_data.dir_table.deinit();
224		self.channel.deinit();
225		self.allocator.destroy(self.channel.buffer_nodes);
226		self.allocator.destroy(self);
227		}
	217	self.allocator.free(self.channel.buffer_nodes);
	218	self.channel.deinit();
	219	self.allocator.destroy(self);
228	220	}
229	221
230	222	pub fn addFile(self: *Self, file_path: []const u8, value: V) !?V {
...	...	@@ -237,217 +229,208 @@ pub fn Watch(comptime V: type) type {
237	229	}
238	230
239	231	fn addFileKEvent(self: *Self, file_path: []const u8, value: V) !?V {
240		const resolved_path = try std.fs.path.resolve(self.allocator, [_][]const u8{file_path});
241		var resolved_path_consumed = false;
242		defer if (!resolved_path_consumed) self.allocator.free(resolved_path);
	232	var realpath_buf: [std.fs.MAX_PATH_BYTES]u8 = undefined;
	233	const realpath = try os.realpath(file_path, &realpath_buf);
243	234
244		var close_op = try CloseOperation.start(self.allocator);
245		var close_op_consumed = false;
246		defer if (!close_op_consumed) close_op.finish();
247
248		const flags = if (comptime std.Target.current.isDarwin()) os.O_SYMLINK \| os.O_EVTONLY else 0;
249		const mode = 0;
250		const fd = try openPosix(self.allocator, resolved_path, flags, mode);
251		close_op.setHandle(fd);
	235	const held = self.os_data.table_lock.acquire();
	236	defer held.release();
252	237
253		var put = try self.allocator.create(OsData.Put);
254		errdefer self.allocator.destroy(put);
255		put.* = OsData.Put{
256		.value = value,
257		.putter_frame = undefined,
258		};
259		put.putter_frame = async self.kqPutEvents(close_op, put);
260		close_op_consumed = true;
261		errdefer {
262		put.cancelled = true;
263		await put.putter_frame;
	238	const gop = try self.os_data.file_table.getOrPut(self.allocator, realpath);
	239	errdefer self.os_data.file_table.removeAssertDiscard(realpath);
	240	if (gop.found_existing) {
	241	const prev_value = gop.entry.value.value;
	242	gop.entry.value.value = value;
	243	return prev_value;
264	244	}
265	245
266		const result = blk: {
267		const held = self.os_data.table_lock.acquire();
268		defer held.release();
269
270		const gop = try self.os_data.file_table.getOrPut(resolved_path);
271		if (gop.found_existing) {
272		const prev_value = gop.kv.value.value;
273		await gop.kv.value.putter_frame;
274		gop.kv.value = put;
275		break :blk prev_value;
276		} else {
277		resolved_path_consumed = true;
278		gop.kv.value = put;
279		break :blk null;
280		}
	246	gop.entry.key = try self.allocator.dupe(u8, realpath);
	247	errdefer self.allocator.free(gop.entry.key);
	248	gop.entry.value = try self.allocator.create(OsData.Put);
	249	errdefer self.allocator.destroy(gop.entry.value);
	250	gop.entry.value.* = .{
	251	.putter_frame = undefined,
	252	.value = value,
281	253	};
282	254
283		return result;
	255	// @TODO Can I close this fd and get an error from bsdWaitKev?
	256	const flags = if (comptime std.Target.current.isDarwin()) os.O_SYMLINK \| os.O_EVTONLY else 0;
	257	const fd = try os.open(realpath, flags, 0);
	258	gop.entry.value.putter_frame = async self.kqPutEvents(fd, gop.entry.key, gop.entry.value);
	259	return null;
284	260	}
285	261
286		fn kqPutEvents(self: Self, close_op: CloseOperation, put: *OsData.Put) void {
	262	fn kqPutEvents(self: Self, fd: os.fd_t, file_path: []const u8, put: OsData.Put) void {
287	263	global_event_loop.beginOneEvent();
288
289	264	defer {
290		close_op.finish();
291	265	global_event_loop.finishOneEvent();
	266	// @TODO: Remove this if we force close otherwise
	267	os.close(fd);
292	268	}
293	269
	270	// We need to manually do a bsdWaitKev to access the fflags.
	271	var resume_node = event.Loop.ResumeNode.Basic{
	272	.base = .{
	273	.id = .Basic,
	274	.handle = @frame(),
	275	.overlapped = event.Loop.ResumeNode.overlapped_init,
	276	},
	277	.kev = undefined,
	278	};
	279
	280	var kevs = [1]os.Kevent{undefined};
	281	const kev = &kevs[0];
	282
294	283	while (!put.cancelled) {
295		if (global_event_loop.bsdWaitKev(
296		@intCast(usize, close_op.getHandle()),
297		os.EVFILT_VNODE,
298		os.NOTE_WRITE \| os.NOTE_DELETE,
299		)) \|kev\| {
300		// TODO handle EV_ERROR
301		if (kev.fflags & os.NOTE_DELETE != 0) {
302		self.channel.put(Self.Event{
303		.id = Event.Id.Delete,
304		.data = put.value,
305		});
306		} else if (kev.fflags & os.NOTE_WRITE != 0) {
307		self.channel.put(Self.Event{
308		.id = Event.Id.CloseWrite,
309		.data = put.value,
310		});
311		}
312		} else \|err\| switch (err) {
313		error.EventNotFound => unreachable,
314		error.ProcessNotFound => unreachable,
315		error.Overflow => unreachable,
316		error.AccessDenied, error.SystemResources => \|casted_err\| {
317		self.channel.put(casted_err);
318		},
	284	kev.* = os.Kevent{
	285	.ident = @intCast(usize, fd),
	286	.filter = os.EVFILT_VNODE,
	287	.flags = os.EV_ADD \| os.EV_ENABLE \| os.EV_CLEAR \| os.EV_ONESHOT \|
	288	os.NOTE_WRITE \| os.NOTE_DELETE \| os.NOTE_REVOKE,
	289	.fflags = 0,
	290	.data = 0,
	291	.udata = @ptrToInt(&resume_node.base),
	292	};
	293	suspend {
	294	global_event_loop.beginOneEvent();
	295	errdefer global_event_loop.finishOneEvent();
	296
	297	const empty_kevs = &[0]os.Kevent{};
	298	_ = os.kevent(global_event_loop.os_data.kqfd, &kevs, empty_kevs, null) catch \|err\| switch (err) {
	299	error.EventNotFound,
	300	error.ProcessNotFound,
	301	error.Overflow,
	302	=> unreachable,
	303	error.AccessDenied, error.SystemResources => \|e\| {
	304	self.channel.put(e);
	305	continue;
	306	},
	307	};
	308	}
	309
	310	if (kev.flags & os.EV_ERROR != 0) {
	311	self.channel.put(os.unexpectedErrno(os.errno(kev.data)));
	312	continue;
	313	}
	314
	315	if (kev.fflags & os.NOTE_DELETE != 0 or kev.fflags & os.NOTE_REVOKE != 0) {
	316	self.channel.put(Self.Event{
	317	.id = .Delete,
	318	.data = put.value,
	319	.dirname = std.fs.path.dirname(file_path) orelse "/",
	320	.basename = std.fs.path.basename(file_path),
	321	});
	322	} else if (kev.fflags & os.NOTE_WRITE != 0) {
	323	self.channel.put(Self.Event{
	324	.id = .CloseWrite,
	325	.data = put.value,
	326	.dirname = std.fs.path.dirname(file_path) orelse "/",
	327	.basename = std.fs.path.basename(file_path),
	328	});
319	329	}
320	330	}
321	331	}
322	332
323	333	fn addFileLinux(self: *Self, file_path: []const u8, value: V) !?V {
324		const dirname = std.fs.path.dirname(file_path) orelse ".";
325		const dirname_with_null = try std.cstr.addNullByte(self.allocator, dirname);
326		var dirname_with_null_consumed = false;
327		defer if (!dirname_with_null_consumed) self.channel.free(dirname_with_null);
328
	334	const dirname = std.fs.path.dirname(file_path) orelse if (file_path[0] == '/') "/" else ".";
329	335	const basename = std.fs.path.basename(file_path);
330		const basename_with_null = try std.cstr.addNullByte(self.allocator, basename);
331		var basename_with_null_consumed = false;
332		defer if (!basename_with_null_consumed) self.allocator.free(basename_with_null);
333	336
334		const wd = try os.inotify_add_watchZ(
	337	const wd = try os.inotify_add_watch(
335	338	self.os_data.inotify_fd,
336		dirname_with_null.ptr,
337		os.linux.IN_CLOSE_WRITE \| os.linux.IN_ONLYDIR \| os.linux.IN_EXCL_UNLINK,
	339	dirname,
	340	os.linux.IN_CLOSE_WRITE \| os.linux.IN_ONLYDIR \| os.linux.IN_DELETE \| os.linux.IN_EXCL_UNLINK,
338	341	);
339	342	// wd is either a newly created watch or an existing one.
340	343
341	344	const held = self.os_data.table_lock.acquire();
342	345	defer held.release();
343	346
344		const gop = try self.os_data.wd_table.getOrPut(wd);
	347	const gop = try self.os_data.wd_table.getOrPut(self.allocator, wd);
	348	errdefer self.os_data.wd_table.removeAssertDiscard(wd);
345	349	if (!gop.found_existing) {
346		gop.kv.value = OsData.Dir{
347		.dirname = dirname_with_null,
	350	gop.entry.value = OsData.Dir{
	351	.dirname = try self.allocator.dupe(u8, dirname),
348	352	.file_table = OsData.FileTable.init(self.allocator),
349	353	};
350		dirname_with_null_consumed = true;
351	354	}
352		const dir = &gop.kv.value;
353	355
354		const file_table_gop = try dir.file_table.getOrPut(basename_with_null);
	356	const dir = &gop.entry.value;
	357	const file_table_gop = try dir.file_table.getOrPut(self.allocator, basename);
	358	errdefer dir.file_table.removeAssertDiscard(basename);
355	359	if (file_table_gop.found_existing) {
356		const prev_value = file_table_gop.kv.value;
357		file_table_gop.kv.value = value;
	360	const prev_value = file_table_gop.entry.value;
	361	file_table_gop.entry.value = value;
358	362	return prev_value;
359	363	} else {
360		file_table_gop.kv.value = value;
361		basename_with_null_consumed = true;
	364	file_table_gop.entry.key = try self.allocator.dupe(u8, basename);
	365	file_table_gop.entry.value = value;
362	366	return null;
363	367	}
364	368	}
365	369
366	370	fn addFileWindows(self: *Self, file_path: []const u8, value: V) !?V {
367	371	// TODO we might need to convert dirname and basename to canonical file paths ("short"?)
368		const dirname = try self.allocator.dupe(u8, std.fs.path.dirname(file_path) orelse ".");
369		var dirname_consumed = false;
370		defer if (!dirname_consumed) self.allocator.free(dirname);
371
372		const dirname_utf16le = try std.unicode.utf8ToUtf16LeWithNull(self.allocator, dirname);
373		defer self.allocator.free(dirname_utf16le);
	372	const dirname = std.fs.path.dirname(file_path) orelse if (file_path[0] == '/') "/" else ".";
	373	var dirname_path_space: windows.PathSpace = undefined;
	374	dirname_path_space.len = try std.unicode.utf8ToUtf16Le(&dirname_path_space.data, dirname);
	375	dirname_path_space.data[dirname_path_space.len] = 0;
374	376
375		// TODO https://github.com/ziglang/zig/issues/265
376	377	const basename = std.fs.path.basename(file_path);
377		const basename_utf16le_null = try std.unicode.utf8ToUtf16LeWithNull(self.allocator, basename);
378		var basename_utf16le_null_consumed = false;
379		defer if (!basename_utf16le_null_consumed) self.allocator.free(basename_utf16le_null);
380		const basename_utf16le_no_null = basename_utf16le_null[0 .. basename_utf16le_null.len - 1];
381
382		const dir_handle = try windows.OpenFile(dirname_utf16le, .{
383		.dir = std.fs.cwd().fd,
384		.access_mask = windows.FILE_LIST_DIRECTORY,
385		.creation = windows.FILE_OPEN,
386		.io_mode = .blocking,
387		.open_dir = true,
388		});
389		var dir_handle_consumed = false;
390		defer if (!dir_handle_consumed) windows.CloseHandle(dir_handle);
	378	var basename_path_space: windows.PathSpace = undefined;
	379	basename_path_space.len = try std.unicode.utf8ToUtf16Le(&basename_path_space.data, basename);
	380	basename_path_space.data[basename_path_space.len] = 0;
391	381
392	382	const held = self.os_data.table_lock.acquire();
393	383	defer held.release();
394	384
395		const gop = try self.os_data.dir_table.getOrPut(dirname);
	385	const gop = try self.os_data.dir_table.getOrPut(self.allocator, dirname);
	386	errdefer self.os_data.dir_table.removeAssertDiscard(dirname);
396	387	if (gop.found_existing) {
397		const dir = gop.kv.value;
398		const held_dir_lock = dir.table_lock.acquire();
399		defer held_dir_lock.release();
	388	const dir = gop.entry.value;
400	389
401		const file_gop = try dir.file_table.getOrPut(basename_utf16le_no_null);
	390	const file_gop = try dir.file_table.getOrPut(self.allocator, basename);
	391	errdefer dir.file_table.removeAssertDiscard(basename);
402	392	if (file_gop.found_existing) {
403		const prev_value = file_gop.kv.value;
404		file_gop.kv.value = value;
	393	const prev_value = file_gop.entry.value;
	394	file_gop.entry.value = value;
405	395	return prev_value;
406	396	} else {
407		file_gop.kv.value = value;
408		basename_utf16le_null_consumed = true;
	397	file_gop.entry.value = value;
	398	file_gop.entry.key = try self.allocator.dupe(u8, basename);
409	399	return null;
410	400	}
411	401	} else {
412		errdefer _ = self.os_data.dir_table.remove(dirname);
	402	const dir_handle = try windows.OpenFile(dirname_path_space.span(), .{
	403	.dir = std.fs.cwd().fd,
	404	.access_mask = windows.FILE_LIST_DIRECTORY,
	405	.creation = windows.FILE_OPEN,
	406	.io_mode = .evented,
	407	.open_dir = true,
	408	});
	409	errdefer windows.CloseHandle(dir_handle);
	410
413	411	const dir = try self.allocator.create(OsData.Dir);
414	412	errdefer self.allocator.destroy(dir);
415	413
	414	gop.entry.key = try self.allocator.dupe(u8, dirname);
	415	errdefer self.allocator.free(gop.entry.key);
	416
416	417	dir.* = OsData.Dir{
417	418	.file_table = OsData.FileTable.init(self.allocator),
418		.table_lock = event.Lock.init(),
419	419	.putter_frame = undefined,
	420	.dir_handle = dir_handle,
420	421	};
421		gop.kv.value = dir;
422		assert((try dir.file_table.put(basename_utf16le_no_null, value)) == null);
423		basename_utf16le_null_consumed = true;
424
425		dir.putter_frame = async self.windowsDirReader(dir_handle, dir);
426		dir_handle_consumed = true;
427
428		dirname_consumed = true;
429
	422	gop.entry.value = dir;
	423	try dir.file_table.put(self.allocator, try self.allocator.dupe(u8, basename), value);
	424	dir.putter_frame = async self.windowsDirReader(dir, gop.entry.key);
430	425	return null;
431	426	}
432	427	}
433	428
434		fn windowsDirReader(self: Self, dir_handle: windows.HANDLE, dir: OsData.Dir) void {
435		self.ref();
436		defer self.deref();
437
438		defer os.close(dir_handle);
439
440		var putter_node = std.atomic.Queue(anyframe).Node{
441		.data = .{ .putter = @frame() },
442		.prev = null,
443		.next = null,
444		};
445		self.os_data.all_putters.put(&putter_node);
446		defer _ = self.os_data.all_putters.remove(&putter_node);
447
	429	fn windowsDirReader(self: Self, dir: OsData.Dir, dirname: []const u8) void {
	430	defer os.close(dir.dir_handle);
448	431	var resume_node = Loop.ResumeNode.Basic{
449	432	.base = Loop.ResumeNode{
450		.id = Loop.ResumeNode.Id.Basic,
	433	.id = .Basic,
451	434	.handle = @frame(),
452	435	.overlapped = windows.OVERLAPPED{
453	436	.Internal = 0,
...	...	@@ -458,157 +441,193 @@ pub fn Watch(comptime V: type) type {
458	441	},
459	442	},
460	443	};
461		var event_buf: [4096]u8 align(@alignOf(windows.FILE_NOTIFY_INFORMATION)) = undefined;
462	444
463		// TODO handle this error not in the channel but in the setup
464		_ = windows.CreateIoCompletionPort(
465		dir_handle,
466		global_event_loop.os_data.io_port,
467		undefined,
468		undefined,
469		) catch \|err\| {
470		self.channel.put(err);
471		return;
472		};
	445	var event_buf: [4096]u8 align(@alignOf(windows.FILE_NOTIFY_INFORMATION)) = undefined;
473	446
474		while (!putter_node.data.cancelled) {
475		{
476		// TODO only 1 beginOneEvent for the whole function
477		global_event_loop.beginOneEvent();
478		errdefer global_event_loop.finishOneEvent();
479		errdefer {
480		_ = windows.kernel32.CancelIoEx(dir_handle, &resume_node.base.overlapped);
481		}
482		suspend {
483		_ = windows.kernel32.ReadDirectoryChangesW(
484		dir_handle,
485		&event_buf,
486		@intCast(windows.DWORD, event_buf.len),
487		windows.FALSE, // watch subtree
488		windows.FILE_NOTIFY_CHANGE_FILE_NAME \| windows.FILE_NOTIFY_CHANGE_DIR_NAME \|
489		windows.FILE_NOTIFY_CHANGE_ATTRIBUTES \| windows.FILE_NOTIFY_CHANGE_SIZE \|
490		windows.FILE_NOTIFY_CHANGE_LAST_WRITE \| windows.FILE_NOTIFY_CHANGE_LAST_ACCESS \|
491		windows.FILE_NOTIFY_CHANGE_CREATION \| windows.FILE_NOTIFY_CHANGE_SECURITY,
492		null, // number of bytes transferred (unused for async)
493		&resume_node.base.overlapped,
494		null, // completion routine - unused because we use IOCP
495		);
496		}
	447	global_event_loop.beginOneEvent();
	448	defer global_event_loop.finishOneEvent();
	449
	450	while (!self.os_data.cancelled) main_loop: {
	451	suspend {
	452	_ = windows.kernel32.ReadDirectoryChangesW(
	453	dir.dir_handle,
	454	&event_buf,
	455	event_buf.len,
	456	windows.FALSE, // watch subtree
	457	windows.FILE_NOTIFY_CHANGE_FILE_NAME \| windows.FILE_NOTIFY_CHANGE_DIR_NAME \|
	458	windows.FILE_NOTIFY_CHANGE_ATTRIBUTES \| windows.FILE_NOTIFY_CHANGE_SIZE \|
	459	windows.FILE_NOTIFY_CHANGE_LAST_WRITE \| windows.FILE_NOTIFY_CHANGE_LAST_ACCESS \|
	460	windows.FILE_NOTIFY_CHANGE_CREATION \| windows.FILE_NOTIFY_CHANGE_SECURITY,
	461	null, // number of bytes transferred (unused for async)
	462	&resume_node.base.overlapped,
	463	null, // completion routine - unused because we use IOCP
	464	);
497	465	}
	466
498	467	var bytes_transferred: windows.DWORD = undefined;
499		if (windows.kernel32.GetOverlappedResult(dir_handle, &resume_node.base.overlapped, &bytes_transferred, windows.FALSE) == 0) {
500		const err = switch (windows.kernel32.GetLastError()) {
	468	if (windows.kernel32.GetOverlappedResult(
	469	dir.dir_handle,
	470	&resume_node.base.overlapped,
	471	&bytes_transferred,
	472	windows.FALSE,
	473	) == 0) {
	474	const potential_error = windows.kernel32.GetLastError();
	475	const err = switch (potential_error) {
	476	.OPERATION_ABORTED, .IO_INCOMPLETE => err_blk: {
	477	if (self.os_data.cancelled)
	478	break :main_loop
	479	else
	480	break :err_blk windows.unexpectedError(potential_error);
	481	},
501	482	else => \|err\| windows.unexpectedError(err),
502	483	};
503	484	self.channel.put(err);
504	485	} else {
505		// can't use @bytesToSlice because of the special variable length name field
506		var ptr = event_buf[0..].ptr;
	486	var ptr: [*]u8 = &event_buf;
507	487	const end_ptr = ptr + bytes_transferred;
508		var ev: *windows.FILE_NOTIFY_INFORMATION = undefined;
509		while (@ptrToInt(ptr) < @ptrToInt(end_ptr)) : (ptr += ev.NextEntryOffset) {
510		ev = @ptrCast(*windows.FILE_NOTIFY_INFORMATION, ptr);
	488	while (@ptrToInt(ptr) < @ptrToInt(end_ptr)) {
	489	const ev = @ptrCast(*const windows.FILE_NOTIFY_INFORMATION, ptr);
511	490	const emit = switch (ev.Action) {
512	491	windows.FILE_ACTION_REMOVED => WatchEventId.Delete,
513		windows.FILE_ACTION_MODIFIED => WatchEventId.CloseWrite,
	492	windows.FILE_ACTION_MODIFIED => .CloseWrite,
514	493	else => null,
515	494	};
516	495	if (emit) \|id\| {
517		const basename_utf16le = ([*]u16)(&ev.FileName)[0 .. ev.FileNameLength / 2];
518		const user_value = blk: {
519		const held = dir.table_lock.acquire();
520		defer held.release();
521
522		if (dir.file_table.get(basename_utf16le)) \|entry\| {
523		break :blk entry.value;
524		} else {
525		break :blk null;
526		}
527		};
528		if (user_value) \|v\| {
	496	const basename_ptr = @ptrCast([*]u16, ptr + @sizeOf(windows.FILE_NOTIFY_INFORMATION));
	497	const basename_utf16le = basename_ptr[0 .. ev.FileNameLength / 2];
	498	var basename_data: [std.fs.MAX_PATH_BYTES]u8 = undefined;
	499	const basename = basename_data[0 .. std.unicode.utf16leToUtf8(&basename_data, basename_utf16le) catch unreachable];
	500
	501	if (dir.file_table.getEntry(basename)) \|entry\| {
529	502	self.channel.put(Event{
530	503	.id = id,
531		.data = v,
	504	.data = entry.value,
	505	.dirname = dirname,
	506	.basename = entry.key,
532	507	});
533	508	}
534	509	}
	510
535	511	if (ev.NextEntryOffset == 0) break;
	512	ptr = @alignCast(@alignOf(windows.FILE_NOTIFY_INFORMATION), ptr + ev.NextEntryOffset);
536	513	}
537	514	}
538	515	}
539	516	}
540	517
541		pub fn removeFile(self: *Self, file_path: []const u8) ?V {
542		@panic("TODO");
	518	pub fn removeFile(self: *Self, file_path: []const u8) !?V {
	519	switch (builtin.os.tag) {
	520	.linux => {
	521	const dirname = std.fs.path.dirname(file_path) orelse if (file_path[0] == '/') "/" else ".";
	522	const basename = std.fs.path.basename(file_path);
	523
	524	const held = self.os_data.table_lock.acquire();
	525	defer held.release();
	526
	527	const dir = self.os_data.wd_table.get(dirname) orelse return null;
	528	if (dir.file_table.remove(basename)) \|file_entry\| {
	529	self.allocator.free(file_entry.key);
	530	return file_entry.value;
	531	}
	532	return null;
	533	},
	534	.windows => {
	535	const dirname = std.fs.path.dirname(file_path) orelse if (file_path[0] == '/') "/" else ".";
	536	const basename = std.fs.path.basename(file_path);
	537
	538	const held = self.os_data.table_lock.acquire();
	539	defer held.release();
	540
	541	const dir = self.os_data.dir_table.get(dirname) orelse return null;
	542	if (dir.file_table.remove(basename)) \|file_entry\| {
	543	self.allocator.free(file_entry.key);
	544	return file_entry.value;
	545	}
	546	return null;
	547	},
	548	.macos, .freebsd, .netbsd, .dragonfly, .openbsd => {
	549	var realpath_buf: [std.fs.MAX_PATH_BYTES]u8 = undefined;
	550	const realpath = try os.realpath(file_path, &realpath_buf);
	551
	552	const held = self.os_data.table_lock.acquire();
	553	defer held.release();
	554
	555	const entry = self.os_data.file_table.get(realpath) orelse return null;
	556	entry.value.cancelled = true;
	557	// @TODO Close the fd here?
	558	await entry.value.putter_frame;
	559	self.allocator.free(entry.key);
	560	self.allocator.destroy(entry.value);
	561
	562	self.os_data.file_table.removeAssertDiscard(realpath);
	563	},
	564	else => @compileError("Unsupported OS"),
	565	}
543	566	}
544	567
545	568	fn linuxEventPutter(self: *Self) void {
546	569	global_event_loop.beginOneEvent();
547	570
548	571	defer {
549		self.os_data.table_lock.deinit();
550		var wd_it = self.os_data.wd_table.iterator();
551		while (wd_it.next()) \|wd_entry\| {
552		var file_it = wd_entry.value.file_table.iterator();
553		while (file_it.next()) \|file_entry\| {
554		self.allocator.free(file_entry.key);
555		}
556		self.allocator.free(wd_entry.value.dirname);
557		wd_entry.value.file_table.deinit();
558		}
559		self.os_data.wd_table.deinit();
560		global_event_loop.finishOneEvent();
	572	std.debug.assert(self.os_data.wd_table.count() == 0);
	573	self.os_data.wd_table.deinit(self.allocator);
561	574	os.close(self.os_data.inotify_fd);
562		self.channel.deinit();
563	575	self.allocator.free(self.channel.buffer_nodes);
	576	self.channel.deinit();
	577	global_event_loop.finishOneEvent();
564	578	}
565	579
566	580	var event_buf: [4096]u8 align(@alignOf(os.linux.inotify_event)) = undefined;
567	581
568	582	while (!self.os_data.cancelled) {
569		const rc = os.linux.read(self.os_data.inotify_fd, &event_buf, event_buf.len);
570		const errno = os.linux.getErrno(rc);
571		switch (errno) {
572		0 => {
573		// can't use @bytesToSlice because of the special variable length name field
574		var ptr = event_buf[0..].ptr;
575		const end_ptr = ptr + event_buf.len;
576		var ev: *os.linux.inotify_event = undefined;
577		while (@ptrToInt(ptr) < @ptrToInt(end_ptr)) {
578		ev = @ptrCast(*os.linux.inotify_event, ptr);
579		if (ev.mask & os.linux.IN_CLOSE_WRITE == os.linux.IN_CLOSE_WRITE) {
580		const basename_ptr = ptr + @sizeOf(os.linux.inotify_event);
581		// `ev.len` counts all bytes in `ev.name` including terminating null byte.
582		const basename_with_null = basename_ptr[0..ev.len];
583		const user_value = blk: {
584		const held = self.os_data.table_lock.acquire();
585		defer held.release();
586
587		const dir = &self.os_data.wd_table.get(ev.wd).?.value;
588		if (dir.file_table.get(basename_with_null)) \|entry\| {
589		break :blk entry.value;
590		} else {
591		break :blk null;
592		}
593		};
594		if (user_value) \|v\| {
595		self.channel.put(Event{
596		.id = WatchEventId.CloseWrite,
597		.data = v,
598		});
599		}
	583	const bytes_read = global_event_loop.read(self.os_data.inotify_fd, &event_buf, false) catch unreachable;
	584
	585	var ptr: [*]u8 = &event_buf;
	586	const end_ptr = ptr + bytes_read;
	587	while (@ptrToInt(ptr) < @ptrToInt(end_ptr)) {
	588	const ev = @ptrCast(*const os.linux.inotify_event, ptr);
	589	if (ev.mask & os.linux.IN_CLOSE_WRITE == os.linux.IN_CLOSE_WRITE) {
	590	const basename_ptr = ptr + @sizeOf(os.linux.inotify_event);
	591	const basename = std.mem.span(@ptrCast([*:0]u8, basename_ptr));
	592
	593	const dir = &self.os_data.wd_table.get(ev.wd).?;
	594	if (dir.file_table.getEntry(basename)) \|file_value\| {
	595	self.channel.put(Event{
	596	.id = .CloseWrite,
	597	.data = file_value.value,
	598	.dirname = dir.dirname,
	599	.basename = file_value.key,
	600	});
	601	}
	602	} else if (ev.mask & os.linux.IN_IGNORED == os.linux.IN_IGNORED) {
	603	// Directory watch was removed
	604	const held = self.os_data.table_lock.acquire();
	605	defer held.release();
	606	if (self.os_data.wd_table.remove(ev.wd)) \|*wd_entry\| {
	607	var file_it = wd_entry.value.file_table.iterator();
	608	while (file_it.next()) \|file_entry\| {
	609	self.allocator.free(file_entry.key);
600	610	}
601
602		ptr = @alignCast(@alignOf(os.linux.inotify_event), ptr + @sizeOf(os.linux.inotify_event) + ev.len);
	611	self.allocator.free(wd_entry.value.dirname);
	612	wd_entry.value.file_table.deinit(self.allocator);
603	613	}
604		},
605		os.linux.EINTR => continue,
606		os.linux.EINVAL => unreachable,
607		os.linux.EFAULT => unreachable,
608		os.linux.EAGAIN => {
609		global_event_loop.linuxWaitFd(self.os_data.inotify_fd, os.linux.EPOLLET \| os.linux.EPOLLIN \| os.EPOLLONESHOT);
610		},
611		else => unreachable,
	614	} else if (ev.mask & os.linux.IN_DELETE == os.linux.IN_DELETE) {
	615	// File or directory was removed or deleted
	616	const basename_ptr = ptr + @sizeOf(os.linux.inotify_event);
	617	const basename = std.mem.span(@ptrCast([*:0]u8, basename_ptr));
	618
	619	const dir = &self.os_data.wd_table.get(ev.wd).?;
	620	if (dir.file_table.getEntry(basename)) \|file_value\| {
	621	self.channel.put(Event{
	622	.id = .Delete,
	623	.data = file_value.value,
	624	.dirname = dir.dirname,
	625	.basename = file_value.key,
	626	});
	627	}
	628	}
	629
	630	ptr = @alignCast(@alignOf(os.linux.inotify_event), ptr + @sizeOf(os.linux.inotify_event) + ev.len);
612	631	}
613	632	}
614	633	}
...	...	@@ -617,19 +636,19 @@ pub fn Watch(comptime V: type) type {
617	636
618	637	const test_tmp_dir = "std_event_fs_test";
619	638
620		test "write a file, watch it, write it again" {
621		// TODO re-enable this test
622		if (true) return error.SkipZigTest;
	639	test "write a file, watch it, write it again, delete it" {
	640	if (!std.io.is_async) return error.SkipZigTest;
	641	// TODO https://github.com/ziglang/zig/issues/1908
	642	if (builtin.single_threaded) return error.SkipZigTest;
623	643
624		try fs.cwd().makePath(test_tmp_dir);
625		defer fs.cwd().deleteTree(test_tmp_dir) catch {};
	644	try std.fs.cwd().makePath(test_tmp_dir);
	645	defer std.fs.cwd().deleteTree(test_tmp_dir) catch {};
626	646
627		const allocator = std.heap.page_allocator;
628		return testFsWatch(&allocator);
	647	return testWriteWatchWriteDelete(std.testing.allocator);
629	648	}
630	649
631		fn testFsWatch(allocator: *Allocator) !void {
632		const file_path = try std.fs.path.join(allocator, [_][]const u8{ test_tmp_dir, "file.txt" });
	650	fn testWriteWatchWriteDelete(allocator: *Allocator) !void {
	651	const file_path = try std.fs.path.join(allocator, &[_][]const u8{ test_tmp_dir, "file.txt" });
633	652	defer allocator.free(file_path);
634	653
635	654	const contents =
...	...	@@ -639,9 +658,10 @@ fn testFsWatch(allocator: *Allocator) !void {
639	658	const line2_offset = 7;
640	659
641	660	// first just write then read the file
642		try writeFile(allocator, file_path, contents);
	661	try std.fs.cwd().writeFile(file_path, contents);
643	662
644		const read_contents = try readFile(allocator, file_path, 1024 * 1024);
	663	const read_contents = try std.fs.cwd().readFileAlloc(allocator, file_path, 1024 * 1024);
	664	defer allocator.free(read_contents);
645	665	testing.expectEqualSlices(u8, contents, read_contents);
646	666
647	667	// now watch the file
...	...	@@ -650,28 +670,49 @@ fn testFsWatch(allocator: *Allocator) !void {
650	670
651	671	testing.expect((try watch.addFile(file_path, {})) == null);
652	672
653		const ev = watch.channel.get();
	673	var ev = async watch.channel.get();
654	674	var ev_consumed = false;
655		defer if (!ev_consumed) await ev;
	675	defer if (!ev_consumed) {
	676	_ = await ev;
	677	};
656	678
657	679	// overwrite line 2
658		const fd = try await openReadWrite(file_path, File.default_mode);
	680	const file = try std.fs.cwd().openFile(file_path, .{ .read = true, .write = true });
659	681	{
660		defer os.close(fd);
661
662		try pwritev(allocator, fd, []const []const u8{"lorem ipsum"}, line2_offset);
	682	defer file.close();
	683	const write_contents = "lorem ipsum";
	684	var iovec = [_]os.iovec_const{.{
	685	.iov_base = write_contents,
	686	.iov_len = write_contents.len,
	687	}};
	688	_ = try file.pwritevAll(&iovec, line2_offset);
663	689	}
664	690
665		ev_consumed = true;
666	691	switch ((try await ev).id) {
667		WatchEventId.CloseWrite => {},
668		WatchEventId.Delete => @panic("wrong event"),
	692	.CloseWrite => {
	693	ev_consumed = true;
	694	},
	695	.Delete => @panic("wrong event"),
669	696	}
670		const contents_updated = try readFile(allocator, file_path, 1024 * 1024);
	697
	698	const contents_updated = try std.fs.cwd().readFileAlloc(allocator, file_path, 1024 * 1024);
	699	defer allocator.free(contents_updated);
	700
671	701	testing.expectEqualSlices(u8,
672	702	\\line 1
673	703	\\lorem ipsum
674	704	, contents_updated);
675	705
676		// TODO test deleting the file and then re-adding it. we should get events for both
	706	ev = async watch.channel.get();
	707	ev_consumed = false;
	708
	709	try std.fs.cwd().deleteFile(file_path);
	710	switch ((try await ev).id) {
	711	.Delete => {
	712	ev_consumed = true;
	713	},
	714	.CloseWrite => @panic("wrong event"),
	715	}
677	716	}
	717
	718	// TODO Test: Add another file watch, remove the old file watch, get an event in the new

lib/std/os/bits/freebsd.zig+3

...	...	@@ -569,6 +569,9 @@ pub const EV_ONESHOT = 0x0010;
569	569	/// clear event state after reporting
570	570	pub const EV_CLEAR = 0x0020;
571	571
	572	/// error, event data contains errno
	573	pub const EV_ERROR = 0x4000;
	574
572	575	/// force immediate event output
573	576	/// ... with or without EV_ERROR
574	577	/// ... use KEVENT_FLAG_ERROR_EVENTS

lib/std/os/windows.zig+6-5

...	...	@@ -109,7 +109,12 @@ pub fn OpenFile(sub_path_w: []const u16, options: OpenFileOptions) OpenError!HAN
109	109	0,
110	110	);
111	111	switch (rc) {
112		.SUCCESS => return result,
	112	.SUCCESS => {
	113	if (std.io.is_async and options.io_mode == .evented) {
	114	_ = CreateIoCompletionPort(result, std.event.Loop.instance.?.os_data.io_port, undefined, undefined) catch undefined;
	115	}
	116	return result;
	117	},
113	118	.OBJECT_NAME_INVALID => unreachable,
114	119	.OBJECT_NAME_NOT_FOUND => return error.FileNotFound,
115	120	.OBJECT_PATH_NOT_FOUND => return error.FileNotFound,
...	...	@@ -418,8 +423,6 @@ pub fn ReadFile(in_hFile: HANDLE, buffer: []u8, offset: ?u64, io_mode: std.io.Mo
418	423	},
419	424	},
420	425	};
421		// TODO only call create io completion port once per fd
422		_ = CreateIoCompletionPort(in_hFile, loop.os_data.io_port, undefined, undefined) catch undefined;
423	426	loop.beginOneEvent();
424	427	suspend {
425	428	// TODO handle buffer bigger than DWORD can hold
...	...	@@ -500,8 +503,6 @@ pub fn WriteFile(
500	503	},
501	504	},
502	505	};
503		// TODO only call create io completion port once per fd
504		_ = CreateIoCompletionPort(handle, loop.os_data.io_port, undefined, undefined) catch undefined;
505	506	loop.beginOneEvent();
506	507	suspend {
507	508	const adjusted_len = math.cast(DWORD, bytes.len) catch maxInt(DWORD);

lib/std/os/windows/bits.zig+2-1

...	...	@@ -813,7 +813,8 @@ pub const FILE_NOTIFY_INFORMATION = extern struct {
813	813	NextEntryOffset: DWORD,
814	814	Action: DWORD,
815	815	FileNameLength: DWORD,
816		FileName: [1]WCHAR,
	816	// Flexible array member
	817	// FileName: [1]WCHAR,
817	818	};
818	819
819	820	pub const FILE_ACTION_ADDED = 0x00000001;

lib/std/os/windows/kernel32.zig+1-1

...	...	@@ -8,7 +8,7 @@ usingnamespace @import("bits.zig");
8	8	pub extern "kernel32" fn AddVectoredExceptionHandler(First: c_ulong, Handler: ?VECTORED_EXCEPTION_HANDLER) callconv(WINAPI) ?*c_void;
9	9	pub extern "kernel32" fn RemoveVectoredExceptionHandler(Handle: HANDLE) callconv(WINAPI) c_ulong;
10	10
11		pub extern "kernel32" fn CancelIoEx(hFile: HANDLE, lpOverlapped: LPOVERLAPPED) callconv(WINAPI) BOOL;
	11	pub extern "kernel32" fn CancelIoEx(hFile: HANDLE, lpOverlapped: ?LPOVERLAPPED) callconv(WINAPI) BOOL;
12	12
13	13	pub extern "kernel32" fn CloseHandle(hObject: HANDLE) callconv(WINAPI) BOOL;
14	14