authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2019-08-08 16:41:38-04:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2019-08-08 16:41:38-04:00
logda56959a9a7dd7b83a8d2bc6b1454ae546a48be6
tree3d069358f719eb604aa527597dbcc3215d4829a9
parent93840f8610974109d129e6940a851c1f7a8c9fce
signature Commit is signed but in an unrecognized format.

closer to std lib event stuff working

3 files changed, 642 insertions(+), 654 deletions(-)

std/event/channel.zig+13-33
......@@ -89,12 +89,7 @@ pub fn Channel(comptime T: type) type {
8989 /// puts a data item in the channel. The promise completes when the value has been added to the
9090 /// buffer, or in the case of a zero size buffer, when the item has been retrieved by a getter.
9191 pub async fn put(self: *SelfChannel, data: T) void {
92 // TODO fix this workaround
93 suspend {
94 resume @handle();
95 }
96
97 var my_tick_node = Loop.NextTickNode.init(@handle());
92 var my_tick_node = Loop.NextTickNode.init(@frame());
9893 var queue_node = std.atomic.Queue(PutNode).Node.init(PutNode{
9994 .tick_node = &my_tick_node,
10095 .data = data,
......@@ -122,15 +117,10 @@ pub fn Channel(comptime T: type) type {
122117 /// await this function to get an item from the channel. If the buffer is empty, the promise will
123118 /// complete when the next item is put in the channel.
124119 pub async fn get(self: *SelfChannel) T {
125 // TODO fix this workaround
126 suspend {
127 resume @handle();
128 }
129
130120 // TODO integrate this function with named return values
131121 // so we can get rid of this extra result copy
132122 var result: T = undefined;
133 var my_tick_node = Loop.NextTickNode.init(@handle());
123 var my_tick_node = Loop.NextTickNode.init(@frame());
134124 var queue_node = std.atomic.Queue(GetNode).Node.init(GetNode{
135125 .tick_node = &my_tick_node,
136126 .data = GetNode.Data{
......@@ -173,15 +163,10 @@ pub fn Channel(comptime T: type) type {
173163 /// Await is necessary for locking purposes. The function will be resumed after checking the channel
174164 /// for data and will not wait for data to be available.
175165 pub async fn getOrNull(self: *SelfChannel) ?T {
176 // TODO fix this workaround
177 suspend {
178 resume @handle();
179 }
180
181166 // TODO integrate this function with named return values
182167 // so we can get rid of this extra result copy
183168 var result: ?T = null;
184 var my_tick_node = Loop.NextTickNode.init(@handle());
169 var my_tick_node = Loop.NextTickNode.init(@frame());
185170 var or_null_node = std.atomic.Queue(*std.atomic.Queue(GetNode).Node).Node.init(undefined);
186171 var queue_node = std.atomic.Queue(GetNode).Node.init(GetNode{
187172 .tick_node = &my_tick_node,
......@@ -334,41 +319,36 @@ test "std.event.Channel" {
334319 const channel = try Channel(i32).create(&loop, 0);
335320 defer channel.destroy();
336321
337 const handle = try async<allocator> testChannelGetter(&loop, channel);
338 defer cancel handle;
339
340 const putter = try async<allocator> testChannelPutter(channel);
341 defer cancel putter;
322 const handle = async testChannelGetter(&loop, channel);
323 const putter = async testChannelPutter(channel);
342324
343325 loop.run();
344326}
345327
346328async fn testChannelGetter(loop: *Loop, channel: *Channel(i32)) void {
347 errdefer @panic("test failed");
348
349 const value1_promise = try async channel.get();
329 const value1_promise = async channel.get();
350330 const value1 = await value1_promise;
351331 testing.expect(value1 == 1234);
352332
353 const value2_promise = try async channel.get();
333 const value2_promise = async channel.get();
354334 const value2 = await value2_promise;
355335 testing.expect(value2 == 4567);
356336
357 const value3_promise = try async channel.getOrNull();
337 const value3_promise = async channel.getOrNull();
358338 const value3 = await value3_promise;
359339 testing.expect(value3 == null);
360340
361 const last_put = try async testPut(channel, 4444);
362 const value4 = await try async channel.getOrNull();
341 const last_put = async testPut(channel, 4444);
342 const value4 = channel.getOrNull();
363343 testing.expect(value4.? == 4444);
364344 await last_put;
365345}
366346
367347async fn testChannelPutter(channel: *Channel(i32)) void {
368 await (async channel.put(1234) catch @panic("out of memory"));
369 await (async channel.put(4567) catch @panic("out of memory"));
348 channel.put(1234);
349 channel.put(4567);
370350}
371351
372352async fn testPut(channel: *Channel(i32), value: i32) void {
373 await (async channel.put(value) catch @panic("out of memory"));
353 channel.put(value);
374354}
std/event/fs.zig+592-592
......@@ -715,594 +715,594 @@ pub const WatchEventId = enum {
715715 Delete,
716716};
717717
718pub const WatchEventError = error{
719 UserResourceLimitReached,
720 SystemResources,
721 AccessDenied,
722 Unexpected, // TODO remove this possibility
723};
724
725pub fn Watch(comptime V: type) type {
726 return struct {
727 channel: *event.Channel(Event.Error!Event),
728 os_data: OsData,
729
730 const OsData = switch (builtin.os) {
731 .macosx, .freebsd, .netbsd => struct {
732 file_table: FileTable,
733 table_lock: event.Lock,
734
735 const FileTable = std.AutoHashMap([]const u8, *Put);
736 const Put = struct {
737 putter: promise,
738 value_ptr: *V,
739 };
740 },
741
742 .linux => LinuxOsData,
743 .windows => WindowsOsData,
744
745 else => @compileError("Unsupported OS"),
746 };
747
748 const WindowsOsData = struct {
749 table_lock: event.Lock,
750 dir_table: DirTable,
751 all_putters: std.atomic.Queue(promise),
752 ref_count: std.atomic.Int(usize),
753
754 const DirTable = std.AutoHashMap([]const u8, *Dir);
755 const FileTable = std.AutoHashMap([]const u16, V);
756
757 const Dir = struct {
758 putter: promise,
759 file_table: FileTable,
760 table_lock: event.Lock,
761 };
762 };
763
764 const LinuxOsData = struct {
765 putter: promise,
766 inotify_fd: i32,
767 wd_table: WdTable,
768 table_lock: event.Lock,
769
770 const WdTable = std.AutoHashMap(i32, Dir);
771 const FileTable = std.AutoHashMap([]const u8, V);
772
773 const Dir = struct {
774 dirname: []const u8,
775 file_table: FileTable,
776 };
777 };
778
779 const FileToHandle = std.AutoHashMap([]const u8, promise);
780
781 const Self = @This();
782
783 pub const Event = struct {
784 id: Id,
785 data: V,
786
787 pub const Id = WatchEventId;
788 pub const Error = WatchEventError;
789 };
790
791 pub fn create(loop: *Loop, event_buf_count: usize) !*Self {
792 const channel = try event.Channel(Self.Event.Error!Self.Event).create(loop, event_buf_count);
793 errdefer channel.destroy();
794
795 switch (builtin.os) {
796 .linux => {
797 const inotify_fd = try os.inotify_init1(os.linux.IN_NONBLOCK | os.linux.IN_CLOEXEC);
798 errdefer os.close(inotify_fd);
799
800 var result: *Self = undefined;
801 _ = try async<loop.allocator> linuxEventPutter(inotify_fd, channel, &result);
802 return result;
803 },
804
805 .windows => {
806 const self = try loop.allocator.create(Self);
807 errdefer loop.allocator.destroy(self);
808 self.* = Self{
809 .channel = channel,
810 .os_data = OsData{
811 .table_lock = event.Lock.init(loop),
812 .dir_table = OsData.DirTable.init(loop.allocator),
813 .ref_count = std.atomic.Int(usize).init(1),
814 .all_putters = std.atomic.Queue(promise).init(),
815 },
816 };
817 return self;
818 },
819
820 .macosx, .freebsd, .netbsd => {
821 const self = try loop.allocator.create(Self);
822 errdefer loop.allocator.destroy(self);
823
824 self.* = Self{
825 .channel = channel,
826 .os_data = OsData{
827 .table_lock = event.Lock.init(loop),
828 .file_table = OsData.FileTable.init(loop.allocator),
829 },
830 };
831 return self;
832 },
833 else => @compileError("Unsupported OS"),
834 }
835 }
836
837 /// All addFile calls and removeFile calls must have completed.
838 pub fn destroy(self: *Self) void {
839 switch (builtin.os) {
840 .macosx, .freebsd, .netbsd => {
841 // TODO we need to cancel the coroutines before destroying the lock
842 self.os_data.table_lock.deinit();
843 var it = self.os_data.file_table.iterator();
844 while (it.next()) |entry| {
845 cancel entry.value.putter;
846 self.channel.loop.allocator.free(entry.key);
847 }
848 self.channel.destroy();
849 },
850 .linux => cancel self.os_data.putter,
851 .windows => {
852 while (self.os_data.all_putters.get()) |putter_node| {
853 cancel putter_node.data;
854 }
855 self.deref();
856 },
857 else => @compileError("Unsupported OS"),
858 }
859 }
860
861 fn ref(self: *Self) void {
862 _ = self.os_data.ref_count.incr();
863 }
864
865 fn deref(self: *Self) void {
866 if (self.os_data.ref_count.decr() == 1) {
867 const allocator = self.channel.loop.allocator;
868 self.os_data.table_lock.deinit();
869 var it = self.os_data.dir_table.iterator();
870 while (it.next()) |entry| {
871 allocator.free(entry.key);
872 allocator.destroy(entry.value);
873 }
874 self.os_data.dir_table.deinit();
875 self.channel.destroy();
876 allocator.destroy(self);
877 }
878 }
879
880 pub async fn addFile(self: *Self, file_path: []const u8, value: V) !?V {
881 switch (builtin.os) {
882 .macosx, .freebsd, .netbsd => return await (async addFileKEvent(self, file_path, value) catch unreachable),
883 .linux => return await (async addFileLinux(self, file_path, value) catch unreachable),
884 .windows => return await (async addFileWindows(self, file_path, value) catch unreachable),
885 else => @compileError("Unsupported OS"),
886 }
887 }
888
889 async fn addFileKEvent(self: *Self, file_path: []const u8, value: V) !?V {
890 const resolved_path = try std.fs.path.resolve(self.channel.loop.allocator, [_][]const u8{file_path});
891 var resolved_path_consumed = false;
892 defer if (!resolved_path_consumed) self.channel.loop.allocator.free(resolved_path);
893
894 var close_op = try CloseOperation.start(self.channel.loop);
895 var close_op_consumed = false;
896 defer if (!close_op_consumed) close_op.finish();
897
898 const flags = if (os.darwin.is_the_target) os.O_SYMLINK | os.O_EVTONLY else 0;
899 const mode = 0;
900 const fd = try await (async openPosix(self.channel.loop, resolved_path, flags, mode) catch unreachable);
901 close_op.setHandle(fd);
902
903 var put_data: *OsData.Put = undefined;
904 const putter = try async self.kqPutEvents(close_op, value, &put_data);
905 close_op_consumed = true;
906 errdefer cancel putter;
907
908 const result = blk: {
909 const held = await (async self.os_data.table_lock.acquire() catch unreachable);
910 defer held.release();
911
912 const gop = try self.os_data.file_table.getOrPut(resolved_path);
913 if (gop.found_existing) {
914 const prev_value = gop.kv.value.value_ptr.*;
915 cancel gop.kv.value.putter;
916 gop.kv.value = put_data;
917 break :blk prev_value;
918 } else {
919 resolved_path_consumed = true;
920 gop.kv.value = put_data;
921 break :blk null;
922 }
923 };
924
925 return result;
926 }
927
928 async fn kqPutEvents(self: *Self, close_op: *CloseOperation, value: V, out_put: **OsData.Put) void {
929 // TODO https://github.com/ziglang/zig/issues/1194
930 suspend {
931 resume @handle();
932 }
933
934 var value_copy = value;
935 var put = OsData.Put{
936 .putter = @handle(),
937 .value_ptr = &value_copy,
938 };
939 out_put.* = &put;
940 self.channel.loop.beginOneEvent();
941
942 defer {
943 close_op.finish();
944 self.channel.loop.finishOneEvent();
945 }
946
947 while (true) {
948 if (await (async self.channel.loop.bsdWaitKev(
949 @intCast(usize, close_op.getHandle()),
950 os.EVFILT_VNODE,
951 os.NOTE_WRITE | os.NOTE_DELETE,
952 ) catch unreachable)) |kev| {
953 // TODO handle EV_ERROR
954 if (kev.fflags & os.NOTE_DELETE != 0) {
955 await (async self.channel.put(Self.Event{
956 .id = Event.Id.Delete,
957 .data = value_copy,
958 }) catch unreachable);
959 } else if (kev.fflags & os.NOTE_WRITE != 0) {
960 await (async self.channel.put(Self.Event{
961 .id = Event.Id.CloseWrite,
962 .data = value_copy,
963 }) catch unreachable);
964 }
965 } else |err| switch (err) {
966 error.EventNotFound => unreachable,
967 error.ProcessNotFound => unreachable,
968 error.Overflow => unreachable,
969 error.AccessDenied, error.SystemResources => |casted_err| {
970 await (async self.channel.put(casted_err) catch unreachable);
971 },
972 }
973 }
974 }
975
976 async fn addFileLinux(self: *Self, file_path: []const u8, value: V) !?V {
977 const value_copy = value;
978
979 const dirname = std.fs.path.dirname(file_path) orelse ".";
980 const dirname_with_null = try std.cstr.addNullByte(self.channel.loop.allocator, dirname);
981 var dirname_with_null_consumed = false;
982 defer if (!dirname_with_null_consumed) self.channel.loop.allocator.free(dirname_with_null);
983
984 const basename = std.fs.path.basename(file_path);
985 const basename_with_null = try std.cstr.addNullByte(self.channel.loop.allocator, basename);
986 var basename_with_null_consumed = false;
987 defer if (!basename_with_null_consumed) self.channel.loop.allocator.free(basename_with_null);
988
989 const wd = try os.inotify_add_watchC(
990 self.os_data.inotify_fd,
991 dirname_with_null.ptr,
992 os.linux.IN_CLOSE_WRITE | os.linux.IN_ONLYDIR | os.linux.IN_EXCL_UNLINK,
993 );
994 // wd is either a newly created watch or an existing one.
995
996 const held = await (async self.os_data.table_lock.acquire() catch unreachable);
997 defer held.release();
998
999 const gop = try self.os_data.wd_table.getOrPut(wd);
1000 if (!gop.found_existing) {
1001 gop.kv.value = OsData.Dir{
1002 .dirname = dirname_with_null,
1003 .file_table = OsData.FileTable.init(self.channel.loop.allocator),
1004 };
1005 dirname_with_null_consumed = true;
1006 }
1007 const dir = &gop.kv.value;
1008
1009 const file_table_gop = try dir.file_table.getOrPut(basename_with_null);
1010 if (file_table_gop.found_existing) {
1011 const prev_value = file_table_gop.kv.value;
1012 file_table_gop.kv.value = value_copy;
1013 return prev_value;
1014 } else {
1015 file_table_gop.kv.value = value_copy;
1016 basename_with_null_consumed = true;
1017 return null;
1018 }
1019 }
1020
1021 async fn addFileWindows(self: *Self, file_path: []const u8, value: V) !?V {
1022 const value_copy = value;
1023 // TODO we might need to convert dirname and basename to canonical file paths ("short"?)
1024
1025 const dirname = try std.mem.dupe(self.channel.loop.allocator, u8, std.fs.path.dirname(file_path) orelse ".");
1026 var dirname_consumed = false;
1027 defer if (!dirname_consumed) self.channel.loop.allocator.free(dirname);
1028
1029 const dirname_utf16le = try std.unicode.utf8ToUtf16LeWithNull(self.channel.loop.allocator, dirname);
1030 defer self.channel.loop.allocator.free(dirname_utf16le);
1031
1032 // TODO https://github.com/ziglang/zig/issues/265
1033 const basename = std.fs.path.basename(file_path);
1034 const basename_utf16le_null = try std.unicode.utf8ToUtf16LeWithNull(self.channel.loop.allocator, basename);
1035 var basename_utf16le_null_consumed = false;
1036 defer if (!basename_utf16le_null_consumed) self.channel.loop.allocator.free(basename_utf16le_null);
1037 const basename_utf16le_no_null = basename_utf16le_null[0 .. basename_utf16le_null.len - 1];
1038
1039 const dir_handle = try windows.CreateFileW(
1040 dirname_utf16le.ptr,
1041 windows.FILE_LIST_DIRECTORY,
1042 windows.FILE_SHARE_READ | windows.FILE_SHARE_DELETE | windows.FILE_SHARE_WRITE,
1043 null,
1044 windows.OPEN_EXISTING,
1045 windows.FILE_FLAG_BACKUP_SEMANTICS | windows.FILE_FLAG_OVERLAPPED,
1046 null,
1047 );
1048 var dir_handle_consumed = false;
1049 defer if (!dir_handle_consumed) windows.CloseHandle(dir_handle);
1050
1051 const held = await (async self.os_data.table_lock.acquire() catch unreachable);
1052 defer held.release();
1053
1054 const gop = try self.os_data.dir_table.getOrPut(dirname);
1055 if (gop.found_existing) {
1056 const dir = gop.kv.value;
1057 const held_dir_lock = await (async dir.table_lock.acquire() catch unreachable);
1058 defer held_dir_lock.release();
1059
1060 const file_gop = try dir.file_table.getOrPut(basename_utf16le_no_null);
1061 if (file_gop.found_existing) {
1062 const prev_value = file_gop.kv.value;
1063 file_gop.kv.value = value_copy;
1064 return prev_value;
1065 } else {
1066 file_gop.kv.value = value_copy;
1067 basename_utf16le_null_consumed = true;
1068 return null;
1069 }
1070 } else {
1071 errdefer _ = self.os_data.dir_table.remove(dirname);
1072 const dir = try self.channel.loop.allocator.create(OsData.Dir);
1073 errdefer self.channel.loop.allocator.destroy(dir);
1074
1075 dir.* = OsData.Dir{
1076 .file_table = OsData.FileTable.init(self.channel.loop.allocator),
1077 .table_lock = event.Lock.init(self.channel.loop),
1078 .putter = undefined,
1079 };
1080 gop.kv.value = dir;
1081 assert((try dir.file_table.put(basename_utf16le_no_null, value_copy)) == null);
1082 basename_utf16le_null_consumed = true;
1083
1084 dir.putter = try async self.windowsDirReader(dir_handle, dir);
1085 dir_handle_consumed = true;
1086
1087 dirname_consumed = true;
1088
1089 return null;
1090 }
1091 }
1092
1093 async fn windowsDirReader(self: *Self, dir_handle: windows.HANDLE, dir: *OsData.Dir) void {
1094 // TODO https://github.com/ziglang/zig/issues/1194
1095 suspend {
1096 resume @handle();
1097 }
1098
1099 self.ref();
1100 defer self.deref();
1101
1102 defer os.close(dir_handle);
1103
1104 var putter_node = std.atomic.Queue(promise).Node{
1105 .data = @handle(),
1106 .prev = null,
1107 .next = null,
1108 };
1109 self.os_data.all_putters.put(&putter_node);
1110 defer _ = self.os_data.all_putters.remove(&putter_node);
1111
1112 var resume_node = Loop.ResumeNode.Basic{
1113 .base = Loop.ResumeNode{
1114 .id = Loop.ResumeNode.Id.Basic,
1115 .handle = @handle(),
1116 .overlapped = windows.OVERLAPPED{
1117 .Internal = 0,
1118 .InternalHigh = 0,
1119 .Offset = 0,
1120 .OffsetHigh = 0,
1121 .hEvent = null,
1122 },
1123 },
1124 };
1125 var event_buf: [4096]u8 align(@alignOf(windows.FILE_NOTIFY_INFORMATION)) = undefined;
1126
1127 // TODO handle this error not in the channel but in the setup
1128 _ = windows.CreateIoCompletionPort(
1129 dir_handle,
1130 self.channel.loop.os_data.io_port,
1131 undefined,
1132 undefined,
1133 ) catch |err| {
1134 await (async self.channel.put(err) catch unreachable);
1135 return;
1136 };
1137
1138 while (true) {
1139 {
1140 // TODO only 1 beginOneEvent for the whole coroutine
1141 self.channel.loop.beginOneEvent();
1142 errdefer self.channel.loop.finishOneEvent();
1143 errdefer {
1144 _ = windows.kernel32.CancelIoEx(dir_handle, &resume_node.base.overlapped);
1145 }
1146 suspend {
1147 _ = windows.kernel32.ReadDirectoryChangesW(
1148 dir_handle,
1149 &event_buf,
1150 @intCast(windows.DWORD, event_buf.len),
1151 windows.FALSE, // watch subtree
1152 windows.FILE_NOTIFY_CHANGE_FILE_NAME | windows.FILE_NOTIFY_CHANGE_DIR_NAME |
1153 windows.FILE_NOTIFY_CHANGE_ATTRIBUTES | windows.FILE_NOTIFY_CHANGE_SIZE |
1154 windows.FILE_NOTIFY_CHANGE_LAST_WRITE | windows.FILE_NOTIFY_CHANGE_LAST_ACCESS |
1155 windows.FILE_NOTIFY_CHANGE_CREATION | windows.FILE_NOTIFY_CHANGE_SECURITY,
1156 null, // number of bytes transferred (unused for async)
1157 &resume_node.base.overlapped,
1158 null, // completion routine - unused because we use IOCP
1159 );
1160 }
1161 }
1162 var bytes_transferred: windows.DWORD = undefined;
1163 if (windows.kernel32.GetOverlappedResult(dir_handle, &resume_node.base.overlapped, &bytes_transferred, windows.FALSE) == 0) {
1164 const err = switch (windows.kernel32.GetLastError()) {
1165 else => |err| windows.unexpectedError(err),
1166 };
1167 await (async self.channel.put(err) catch unreachable);
1168 } else {
1169 // can't use @bytesToSlice because of the special variable length name field
1170 var ptr = event_buf[0..].ptr;
1171 const end_ptr = ptr + bytes_transferred;
1172 var ev: *windows.FILE_NOTIFY_INFORMATION = undefined;
1173 while (@ptrToInt(ptr) < @ptrToInt(end_ptr)) : (ptr += ev.NextEntryOffset) {
1174 ev = @ptrCast(*windows.FILE_NOTIFY_INFORMATION, ptr);
1175 const emit = switch (ev.Action) {
1176 windows.FILE_ACTION_REMOVED => WatchEventId.Delete,
1177 windows.FILE_ACTION_MODIFIED => WatchEventId.CloseWrite,
1178 else => null,
1179 };
1180 if (emit) |id| {
1181 const basename_utf16le = ([*]u16)(&ev.FileName)[0 .. ev.FileNameLength / 2];
1182 const user_value = blk: {
1183 const held = await (async dir.table_lock.acquire() catch unreachable);
1184 defer held.release();
1185
1186 if (dir.file_table.get(basename_utf16le)) |entry| {
1187 break :blk entry.value;
1188 } else {
1189 break :blk null;
1190 }
1191 };
1192 if (user_value) |v| {
1193 await (async self.channel.put(Event{
1194 .id = id,
1195 .data = v,
1196 }) catch unreachable);
1197 }
1198 }
1199 if (ev.NextEntryOffset == 0) break;
1200 }
1201 }
1202 }
1203 }
1204
1205 pub async fn removeFile(self: *Self, file_path: []const u8) ?V {
1206 @panic("TODO");
1207 }
1208
1209 async fn linuxEventPutter(inotify_fd: i32, channel: *event.Channel(Event.Error!Event), out_watch: **Self) void {
1210 // TODO https://github.com/ziglang/zig/issues/1194
1211 suspend {
1212 resume @handle();
1213 }
1214
1215 const loop = channel.loop;
1216
1217 var watch = Self{
1218 .channel = channel,
1219 .os_data = OsData{
1220 .putter = @handle(),
1221 .inotify_fd = inotify_fd,
1222 .wd_table = OsData.WdTable.init(loop.allocator),
1223 .table_lock = event.Lock.init(loop),
1224 },
1225 };
1226 out_watch.* = &watch;
1227
1228 loop.beginOneEvent();
1229
1230 defer {
1231 watch.os_data.table_lock.deinit();
1232 var wd_it = watch.os_data.wd_table.iterator();
1233 while (wd_it.next()) |wd_entry| {
1234 var file_it = wd_entry.value.file_table.iterator();
1235 while (file_it.next()) |file_entry| {
1236 loop.allocator.free(file_entry.key);
1237 }
1238 loop.allocator.free(wd_entry.value.dirname);
1239 }
1240 loop.finishOneEvent();
1241 os.close(inotify_fd);
1242 channel.destroy();
1243 }
1244
1245 var event_buf: [4096]u8 align(@alignOf(os.linux.inotify_event)) = undefined;
1246
1247 while (true) {
1248 const rc = os.linux.read(inotify_fd, &event_buf, event_buf.len);
1249 const errno = os.linux.getErrno(rc);
1250 switch (errno) {
1251 0 => {
1252 // can't use @bytesToSlice because of the special variable length name field
1253 var ptr = event_buf[0..].ptr;
1254 const end_ptr = ptr + event_buf.len;
1255 var ev: *os.linux.inotify_event = undefined;
1256 while (@ptrToInt(ptr) < @ptrToInt(end_ptr)) : (ptr += @sizeOf(os.linux.inotify_event) + ev.len) {
1257 ev = @ptrCast(*os.linux.inotify_event, ptr);
1258 if (ev.mask & os.linux.IN_CLOSE_WRITE == os.linux.IN_CLOSE_WRITE) {
1259 const basename_ptr = ptr + @sizeOf(os.linux.inotify_event);
1260 const basename_with_null = basename_ptr[0 .. std.mem.len(u8, basename_ptr) + 1];
1261 const user_value = blk: {
1262 const held = await (async watch.os_data.table_lock.acquire() catch unreachable);
1263 defer held.release();
1264
1265 const dir = &watch.os_data.wd_table.get(ev.wd).?.value;
1266 if (dir.file_table.get(basename_with_null)) |entry| {
1267 break :blk entry.value;
1268 } else {
1269 break :blk null;
1270 }
1271 };
1272 if (user_value) |v| {
1273 await (async channel.put(Event{
1274 .id = WatchEventId.CloseWrite,
1275 .data = v,
1276 }) catch unreachable);
1277 }
1278 }
1279 }
1280 },
1281 os.linux.EINTR => continue,
1282 os.linux.EINVAL => unreachable,
1283 os.linux.EFAULT => unreachable,
1284 os.linux.EAGAIN => {
1285 (await (async loop.linuxWaitFd(
1286 inotify_fd,
1287 os.linux.EPOLLET | os.linux.EPOLLIN,
1288 ) catch unreachable)) catch |err| {
1289 const transformed_err = switch (err) {
1290 error.FileDescriptorAlreadyPresentInSet => unreachable,
1291 error.OperationCausesCircularLoop => unreachable,
1292 error.FileDescriptorNotRegistered => unreachable,
1293 error.FileDescriptorIncompatibleWithEpoll => unreachable,
1294 error.Unexpected => unreachable,
1295 else => |e| e,
1296 };
1297 await (async channel.put(transformed_err) catch unreachable);
1298 };
1299 },
1300 else => unreachable,
1301 }
1302 }
1303 }
1304 };
1305}
718//pub const WatchEventError = error{
719// UserResourceLimitReached,
720// SystemResources,
721// AccessDenied,
722// Unexpected, // TODO remove this possibility
723//};
724//
725//pub fn Watch(comptime V: type) type {
726// return struct {
727// channel: *event.Channel(Event.Error!Event),
728// os_data: OsData,
729//
730// const OsData = switch (builtin.os) {
731// .macosx, .freebsd, .netbsd => struct {
732// file_table: FileTable,
733// table_lock: event.Lock,
734//
735// const FileTable = std.AutoHashMap([]const u8, *Put);
736// const Put = struct {
737// putter: promise,
738// value_ptr: *V,
739// };
740// },
741//
742// .linux => LinuxOsData,
743// .windows => WindowsOsData,
744//
745// else => @compileError("Unsupported OS"),
746// };
747//
748// const WindowsOsData = struct {
749// table_lock: event.Lock,
750// dir_table: DirTable,
751// all_putters: std.atomic.Queue(promise),
752// ref_count: std.atomic.Int(usize),
753//
754// const DirTable = std.AutoHashMap([]const u8, *Dir);
755// const FileTable = std.AutoHashMap([]const u16, V);
756//
757// const Dir = struct {
758// putter: promise,
759// file_table: FileTable,
760// table_lock: event.Lock,
761// };
762// };
763//
764// const LinuxOsData = struct {
765// putter: promise,
766// inotify_fd: i32,
767// wd_table: WdTable,
768// table_lock: event.Lock,
769//
770// const WdTable = std.AutoHashMap(i32, Dir);
771// const FileTable = std.AutoHashMap([]const u8, V);
772//
773// const Dir = struct {
774// dirname: []const u8,
775// file_table: FileTable,
776// };
777// };
778//
779// const FileToHandle = std.AutoHashMap([]const u8, promise);
780//
781// const Self = @This();
782//
783// pub const Event = struct {
784// id: Id,
785// data: V,
786//
787// pub const Id = WatchEventId;
788// pub const Error = WatchEventError;
789// };
790//
791// pub fn create(loop: *Loop, event_buf_count: usize) !*Self {
792// const channel = try event.Channel(Self.Event.Error!Self.Event).create(loop, event_buf_count);
793// errdefer channel.destroy();
794//
795// switch (builtin.os) {
796// .linux => {
797// const inotify_fd = try os.inotify_init1(os.linux.IN_NONBLOCK | os.linux.IN_CLOEXEC);
798// errdefer os.close(inotify_fd);
799//
800// var result: *Self = undefined;
801// _ = try async<loop.allocator> linuxEventPutter(inotify_fd, channel, &result);
802// return result;
803// },
804//
805// .windows => {
806// const self = try loop.allocator.create(Self);
807// errdefer loop.allocator.destroy(self);
808// self.* = Self{
809// .channel = channel,
810// .os_data = OsData{
811// .table_lock = event.Lock.init(loop),
812// .dir_table = OsData.DirTable.init(loop.allocator),
813// .ref_count = std.atomic.Int(usize).init(1),
814// .all_putters = std.atomic.Queue(promise).init(),
815// },
816// };
817// return self;
818// },
819//
820// .macosx, .freebsd, .netbsd => {
821// const self = try loop.allocator.create(Self);
822// errdefer loop.allocator.destroy(self);
823//
824// self.* = Self{
825// .channel = channel,
826// .os_data = OsData{
827// .table_lock = event.Lock.init(loop),
828// .file_table = OsData.FileTable.init(loop.allocator),
829// },
830// };
831// return self;
832// },
833// else => @compileError("Unsupported OS"),
834// }
835// }
836//
837// /// All addFile calls and removeFile calls must have completed.
838// pub fn destroy(self: *Self) void {
839// switch (builtin.os) {
840// .macosx, .freebsd, .netbsd => {
841// // TODO we need to cancel the coroutines before destroying the lock
842// self.os_data.table_lock.deinit();
843// var it = self.os_data.file_table.iterator();
844// while (it.next()) |entry| {
845// cancel entry.value.putter;
846// self.channel.loop.allocator.free(entry.key);
847// }
848// self.channel.destroy();
849// },
850// .linux => cancel self.os_data.putter,
851// .windows => {
852// while (self.os_data.all_putters.get()) |putter_node| {
853// cancel putter_node.data;
854// }
855// self.deref();
856// },
857// else => @compileError("Unsupported OS"),
858// }
859// }
860//
861// fn ref(self: *Self) void {
862// _ = self.os_data.ref_count.incr();
863// }
864//
865// fn deref(self: *Self) void {
866// if (self.os_data.ref_count.decr() == 1) {
867// const allocator = self.channel.loop.allocator;
868// self.os_data.table_lock.deinit();
869// var it = self.os_data.dir_table.iterator();
870// while (it.next()) |entry| {
871// allocator.free(entry.key);
872// allocator.destroy(entry.value);
873// }
874// self.os_data.dir_table.deinit();
875// self.channel.destroy();
876// allocator.destroy(self);
877// }
878// }
879//
880// pub async fn addFile(self: *Self, file_path: []const u8, value: V) !?V {
881// switch (builtin.os) {
882// .macosx, .freebsd, .netbsd => return await (async addFileKEvent(self, file_path, value) catch unreachable),
883// .linux => return await (async addFileLinux(self, file_path, value) catch unreachable),
884// .windows => return await (async addFileWindows(self, file_path, value) catch unreachable),
885// else => @compileError("Unsupported OS"),
886// }
887// }
888//
889// async fn addFileKEvent(self: *Self, file_path: []const u8, value: V) !?V {
890// const resolved_path = try std.fs.path.resolve(self.channel.loop.allocator, [_][]const u8{file_path});
891// var resolved_path_consumed = false;
892// defer if (!resolved_path_consumed) self.channel.loop.allocator.free(resolved_path);
893//
894// var close_op = try CloseOperation.start(self.channel.loop);
895// var close_op_consumed = false;
896// defer if (!close_op_consumed) close_op.finish();
897//
898// const flags = if (os.darwin.is_the_target) os.O_SYMLINK | os.O_EVTONLY else 0;
899// const mode = 0;
900// const fd = try await (async openPosix(self.channel.loop, resolved_path, flags, mode) catch unreachable);
901// close_op.setHandle(fd);
902//
903// var put_data: *OsData.Put = undefined;
904// const putter = try async self.kqPutEvents(close_op, value, &put_data);
905// close_op_consumed = true;
906// errdefer cancel putter;
907//
908// const result = blk: {
909// const held = await (async self.os_data.table_lock.acquire() catch unreachable);
910// defer held.release();
911//
912// const gop = try self.os_data.file_table.getOrPut(resolved_path);
913// if (gop.found_existing) {
914// const prev_value = gop.kv.value.value_ptr.*;
915// cancel gop.kv.value.putter;
916// gop.kv.value = put_data;
917// break :blk prev_value;
918// } else {
919// resolved_path_consumed = true;
920// gop.kv.value = put_data;
921// break :blk null;
922// }
923// };
924//
925// return result;
926// }
927//
928// async fn kqPutEvents(self: *Self, close_op: *CloseOperation, value: V, out_put: **OsData.Put) void {
929// // TODO https://github.com/ziglang/zig/issues/1194
930// suspend {
931// resume @handle();
932// }
933//
934// var value_copy = value;
935// var put = OsData.Put{
936// .putter = @handle(),
937// .value_ptr = &value_copy,
938// };
939// out_put.* = &put;
940// self.channel.loop.beginOneEvent();
941//
942// defer {
943// close_op.finish();
944// self.channel.loop.finishOneEvent();
945// }
946//
947// while (true) {
948// if (await (async self.channel.loop.bsdWaitKev(
949// @intCast(usize, close_op.getHandle()),
950// os.EVFILT_VNODE,
951// os.NOTE_WRITE | os.NOTE_DELETE,
952// ) catch unreachable)) |kev| {
953// // TODO handle EV_ERROR
954// if (kev.fflags & os.NOTE_DELETE != 0) {
955// await (async self.channel.put(Self.Event{
956// .id = Event.Id.Delete,
957// .data = value_copy,
958// }) catch unreachable);
959// } else if (kev.fflags & os.NOTE_WRITE != 0) {
960// await (async self.channel.put(Self.Event{
961// .id = Event.Id.CloseWrite,
962// .data = value_copy,
963// }) catch unreachable);
964// }
965// } else |err| switch (err) {
966// error.EventNotFound => unreachable,
967// error.ProcessNotFound => unreachable,
968// error.Overflow => unreachable,
969// error.AccessDenied, error.SystemResources => |casted_err| {
970// await (async self.channel.put(casted_err) catch unreachable);
971// },
972// }
973// }
974// }
975//
976// async fn addFileLinux(self: *Self, file_path: []const u8, value: V) !?V {
977// const value_copy = value;
978//
979// const dirname = std.fs.path.dirname(file_path) orelse ".";
980// const dirname_with_null = try std.cstr.addNullByte(self.channel.loop.allocator, dirname);
981// var dirname_with_null_consumed = false;
982// defer if (!dirname_with_null_consumed) self.channel.loop.allocator.free(dirname_with_null);
983//
984// const basename = std.fs.path.basename(file_path);
985// const basename_with_null = try std.cstr.addNullByte(self.channel.loop.allocator, basename);
986// var basename_with_null_consumed = false;
987// defer if (!basename_with_null_consumed) self.channel.loop.allocator.free(basename_with_null);
988//
989// const wd = try os.inotify_add_watchC(
990// self.os_data.inotify_fd,
991// dirname_with_null.ptr,
992// os.linux.IN_CLOSE_WRITE | os.linux.IN_ONLYDIR | os.linux.IN_EXCL_UNLINK,
993// );
994// // wd is either a newly created watch or an existing one.
995//
996// const held = await (async self.os_data.table_lock.acquire() catch unreachable);
997// defer held.release();
998//
999// const gop = try self.os_data.wd_table.getOrPut(wd);
1000// if (!gop.found_existing) {
1001// gop.kv.value = OsData.Dir{
1002// .dirname = dirname_with_null,
1003// .file_table = OsData.FileTable.init(self.channel.loop.allocator),
1004// };
1005// dirname_with_null_consumed = true;
1006// }
1007// const dir = &gop.kv.value;
1008//
1009// const file_table_gop = try dir.file_table.getOrPut(basename_with_null);
1010// if (file_table_gop.found_existing) {
1011// const prev_value = file_table_gop.kv.value;
1012// file_table_gop.kv.value = value_copy;
1013// return prev_value;
1014// } else {
1015// file_table_gop.kv.value = value_copy;
1016// basename_with_null_consumed = true;
1017// return null;
1018// }
1019// }
1020//
1021// async fn addFileWindows(self: *Self, file_path: []const u8, value: V) !?V {
1022// const value_copy = value;
1023// // TODO we might need to convert dirname and basename to canonical file paths ("short"?)
1024//
1025// const dirname = try std.mem.dupe(self.channel.loop.allocator, u8, std.fs.path.dirname(file_path) orelse ".");
1026// var dirname_consumed = false;
1027// defer if (!dirname_consumed) self.channel.loop.allocator.free(dirname);
1028//
1029// const dirname_utf16le = try std.unicode.utf8ToUtf16LeWithNull(self.channel.loop.allocator, dirname);
1030// defer self.channel.loop.allocator.free(dirname_utf16le);
1031//
1032// // TODO https://github.com/ziglang/zig/issues/265
1033// const basename = std.fs.path.basename(file_path);
1034// const basename_utf16le_null = try std.unicode.utf8ToUtf16LeWithNull(self.channel.loop.allocator, basename);
1035// var basename_utf16le_null_consumed = false;
1036// defer if (!basename_utf16le_null_consumed) self.channel.loop.allocator.free(basename_utf16le_null);
1037// const basename_utf16le_no_null = basename_utf16le_null[0 .. basename_utf16le_null.len - 1];
1038//
1039// const dir_handle = try windows.CreateFileW(
1040// dirname_utf16le.ptr,
1041// windows.FILE_LIST_DIRECTORY,
1042// windows.FILE_SHARE_READ | windows.FILE_SHARE_DELETE | windows.FILE_SHARE_WRITE,
1043// null,
1044// windows.OPEN_EXISTING,
1045// windows.FILE_FLAG_BACKUP_SEMANTICS | windows.FILE_FLAG_OVERLAPPED,
1046// null,
1047// );
1048// var dir_handle_consumed = false;
1049// defer if (!dir_handle_consumed) windows.CloseHandle(dir_handle);
1050//
1051// const held = await (async self.os_data.table_lock.acquire() catch unreachable);
1052// defer held.release();
1053//
1054// const gop = try self.os_data.dir_table.getOrPut(dirname);
1055// if (gop.found_existing) {
1056// const dir = gop.kv.value;
1057// const held_dir_lock = await (async dir.table_lock.acquire() catch unreachable);
1058// defer held_dir_lock.release();
1059//
1060// const file_gop = try dir.file_table.getOrPut(basename_utf16le_no_null);
1061// if (file_gop.found_existing) {
1062// const prev_value = file_gop.kv.value;
1063// file_gop.kv.value = value_copy;
1064// return prev_value;
1065// } else {
1066// file_gop.kv.value = value_copy;
1067// basename_utf16le_null_consumed = true;
1068// return null;
1069// }
1070// } else {
1071// errdefer _ = self.os_data.dir_table.remove(dirname);
1072// const dir = try self.channel.loop.allocator.create(OsData.Dir);
1073// errdefer self.channel.loop.allocator.destroy(dir);
1074//
1075// dir.* = OsData.Dir{
1076// .file_table = OsData.FileTable.init(self.channel.loop.allocator),
1077// .table_lock = event.Lock.init(self.channel.loop),
1078// .putter = undefined,
1079// };
1080// gop.kv.value = dir;
1081// assert((try dir.file_table.put(basename_utf16le_no_null, value_copy)) == null);
1082// basename_utf16le_null_consumed = true;
1083//
1084// dir.putter = try async self.windowsDirReader(dir_handle, dir);
1085// dir_handle_consumed = true;
1086//
1087// dirname_consumed = true;
1088//
1089// return null;
1090// }
1091// }
1092//
1093// async fn windowsDirReader(self: *Self, dir_handle: windows.HANDLE, dir: *OsData.Dir) void {
1094// // TODO https://github.com/ziglang/zig/issues/1194
1095// suspend {
1096// resume @handle();
1097// }
1098//
1099// self.ref();
1100// defer self.deref();
1101//
1102// defer os.close(dir_handle);
1103//
1104// var putter_node = std.atomic.Queue(promise).Node{
1105// .data = @handle(),
1106// .prev = null,
1107// .next = null,
1108// };
1109// self.os_data.all_putters.put(&putter_node);
1110// defer _ = self.os_data.all_putters.remove(&putter_node);
1111//
1112// var resume_node = Loop.ResumeNode.Basic{
1113// .base = Loop.ResumeNode{
1114// .id = Loop.ResumeNode.Id.Basic,
1115// .handle = @handle(),
1116// .overlapped = windows.OVERLAPPED{
1117// .Internal = 0,
1118// .InternalHigh = 0,
1119// .Offset = 0,
1120// .OffsetHigh = 0,
1121// .hEvent = null,
1122// },
1123// },
1124// };
1125// var event_buf: [4096]u8 align(@alignOf(windows.FILE_NOTIFY_INFORMATION)) = undefined;
1126//
1127// // TODO handle this error not in the channel but in the setup
1128// _ = windows.CreateIoCompletionPort(
1129// dir_handle,
1130// self.channel.loop.os_data.io_port,
1131// undefined,
1132// undefined,
1133// ) catch |err| {
1134// await (async self.channel.put(err) catch unreachable);
1135// return;
1136// };
1137//
1138// while (true) {
1139// {
1140// // TODO only 1 beginOneEvent for the whole coroutine
1141// self.channel.loop.beginOneEvent();
1142// errdefer self.channel.loop.finishOneEvent();
1143// errdefer {
1144// _ = windows.kernel32.CancelIoEx(dir_handle, &resume_node.base.overlapped);
1145// }
1146// suspend {
1147// _ = windows.kernel32.ReadDirectoryChangesW(
1148// dir_handle,
1149// &event_buf,
1150// @intCast(windows.DWORD, event_buf.len),
1151// windows.FALSE, // watch subtree
1152// windows.FILE_NOTIFY_CHANGE_FILE_NAME | windows.FILE_NOTIFY_CHANGE_DIR_NAME |
1153// windows.FILE_NOTIFY_CHANGE_ATTRIBUTES | windows.FILE_NOTIFY_CHANGE_SIZE |
1154// windows.FILE_NOTIFY_CHANGE_LAST_WRITE | windows.FILE_NOTIFY_CHANGE_LAST_ACCESS |
1155// windows.FILE_NOTIFY_CHANGE_CREATION | windows.FILE_NOTIFY_CHANGE_SECURITY,
1156// null, // number of bytes transferred (unused for async)
1157// &resume_node.base.overlapped,
1158// null, // completion routine - unused because we use IOCP
1159// );
1160// }
1161// }
1162// var bytes_transferred: windows.DWORD = undefined;
1163// if (windows.kernel32.GetOverlappedResult(dir_handle, &resume_node.base.overlapped, &bytes_transferred, windows.FALSE) == 0) {
1164// const err = switch (windows.kernel32.GetLastError()) {
1165// else => |err| windows.unexpectedError(err),
1166// };
1167// await (async self.channel.put(err) catch unreachable);
1168// } else {
1169// // can't use @bytesToSlice because of the special variable length name field
1170// var ptr = event_buf[0..].ptr;
1171// const end_ptr = ptr + bytes_transferred;
1172// var ev: *windows.FILE_NOTIFY_INFORMATION = undefined;
1173// while (@ptrToInt(ptr) < @ptrToInt(end_ptr)) : (ptr += ev.NextEntryOffset) {
1174// ev = @ptrCast(*windows.FILE_NOTIFY_INFORMATION, ptr);
1175// const emit = switch (ev.Action) {
1176// windows.FILE_ACTION_REMOVED => WatchEventId.Delete,
1177// windows.FILE_ACTION_MODIFIED => WatchEventId.CloseWrite,
1178// else => null,
1179// };
1180// if (emit) |id| {
1181// const basename_utf16le = ([*]u16)(&ev.FileName)[0 .. ev.FileNameLength / 2];
1182// const user_value = blk: {
1183// const held = await (async dir.table_lock.acquire() catch unreachable);
1184// defer held.release();
1185//
1186// if (dir.file_table.get(basename_utf16le)) |entry| {
1187// break :blk entry.value;
1188// } else {
1189// break :blk null;
1190// }
1191// };
1192// if (user_value) |v| {
1193// await (async self.channel.put(Event{
1194// .id = id,
1195// .data = v,
1196// }) catch unreachable);
1197// }
1198// }
1199// if (ev.NextEntryOffset == 0) break;
1200// }
1201// }
1202// }
1203// }
1204//
1205// pub async fn removeFile(self: *Self, file_path: []const u8) ?V {
1206// @panic("TODO");
1207// }
1208//
1209// async fn linuxEventPutter(inotify_fd: i32, channel: *event.Channel(Event.Error!Event), out_watch: **Self) void {
1210// // TODO https://github.com/ziglang/zig/issues/1194
1211// suspend {
1212// resume @handle();
1213// }
1214//
1215// const loop = channel.loop;
1216//
1217// var watch = Self{
1218// .channel = channel,
1219// .os_data = OsData{
1220// .putter = @handle(),
1221// .inotify_fd = inotify_fd,
1222// .wd_table = OsData.WdTable.init(loop.allocator),
1223// .table_lock = event.Lock.init(loop),
1224// },
1225// };
1226// out_watch.* = &watch;
1227//
1228// loop.beginOneEvent();
1229//
1230// defer {
1231// watch.os_data.table_lock.deinit();
1232// var wd_it = watch.os_data.wd_table.iterator();
1233// while (wd_it.next()) |wd_entry| {
1234// var file_it = wd_entry.value.file_table.iterator();
1235// while (file_it.next()) |file_entry| {
1236// loop.allocator.free(file_entry.key);
1237// }
1238// loop.allocator.free(wd_entry.value.dirname);
1239// }
1240// loop.finishOneEvent();
1241// os.close(inotify_fd);
1242// channel.destroy();
1243// }
1244//
1245// var event_buf: [4096]u8 align(@alignOf(os.linux.inotify_event)) = undefined;
1246//
1247// while (true) {
1248// const rc = os.linux.read(inotify_fd, &event_buf, event_buf.len);
1249// const errno = os.linux.getErrno(rc);
1250// switch (errno) {
1251// 0 => {
1252// // can't use @bytesToSlice because of the special variable length name field
1253// var ptr = event_buf[0..].ptr;
1254// const end_ptr = ptr + event_buf.len;
1255// var ev: *os.linux.inotify_event = undefined;
1256// while (@ptrToInt(ptr) < @ptrToInt(end_ptr)) : (ptr += @sizeOf(os.linux.inotify_event) + ev.len) {
1257// ev = @ptrCast(*os.linux.inotify_event, ptr);
1258// if (ev.mask & os.linux.IN_CLOSE_WRITE == os.linux.IN_CLOSE_WRITE) {
1259// const basename_ptr = ptr + @sizeOf(os.linux.inotify_event);
1260// const basename_with_null = basename_ptr[0 .. std.mem.len(u8, basename_ptr) + 1];
1261// const user_value = blk: {
1262// const held = await (async watch.os_data.table_lock.acquire() catch unreachable);
1263// defer held.release();
1264//
1265// const dir = &watch.os_data.wd_table.get(ev.wd).?.value;
1266// if (dir.file_table.get(basename_with_null)) |entry| {
1267// break :blk entry.value;
1268// } else {
1269// break :blk null;
1270// }
1271// };
1272// if (user_value) |v| {
1273// await (async channel.put(Event{
1274// .id = WatchEventId.CloseWrite,
1275// .data = v,
1276// }) catch unreachable);
1277// }
1278// }
1279// }
1280// },
1281// os.linux.EINTR => continue,
1282// os.linux.EINVAL => unreachable,
1283// os.linux.EFAULT => unreachable,
1284// os.linux.EAGAIN => {
1285// (await (async loop.linuxWaitFd(
1286// inotify_fd,
1287// os.linux.EPOLLET | os.linux.EPOLLIN,
1288// ) catch unreachable)) catch |err| {
1289// const transformed_err = switch (err) {
1290// error.FileDescriptorAlreadyPresentInSet => unreachable,
1291// error.OperationCausesCircularLoop => unreachable,
1292// error.FileDescriptorNotRegistered => unreachable,
1293// error.FileDescriptorIncompatibleWithEpoll => unreachable,
1294// error.Unexpected => unreachable,
1295// else => |e| e,
1296// };
1297// await (async channel.put(transformed_err) catch unreachable);
1298// };
1299// },
1300// else => unreachable,
1301// }
1302// }
1303// }
1304// };
1305//}
13061306
13071307const test_tmp_dir = "std_event_fs_test";
13081308
......@@ -1397,11 +1397,11 @@ pub const OutStream = struct {
13971397 };
13981398 }
13991399
1400 async<*mem.Allocator> fn writeFn(out_stream: *Stream, bytes: []const u8) Error!void {
1400 fn writeFn(out_stream: *Stream, bytes: []const u8) Error!void {
14011401 const self = @fieldParentPtr(OutStream, "stream", out_stream);
14021402 const offset = self.offset;
14031403 self.offset += bytes.len;
1404 return await (async pwritev(self.loop, self.fd, [][]const u8{bytes}, offset) catch unreachable);
1404 return pwritev(self.loop, self.fd, [][]const u8{bytes}, offset);
14051405 }
14061406};
14071407
......@@ -1423,9 +1423,9 @@ pub const InStream = struct {
14231423 };
14241424 }
14251425
1426 async<*mem.Allocator> fn readFn(in_stream: *Stream, bytes: []u8) Error!usize {
1426 fn readFn(in_stream: *Stream, bytes: []u8) Error!usize {
14271427 const self = @fieldParentPtr(InStream, "stream", in_stream);
1428 const amt = try await (async preadv(self.loop, self.fd, [][]u8{bytes}, self.offset) catch unreachable);
1428 const amt = try preadv(self.loop, self.fd, [][]u8{bytes}, self.offset);
14291429 self.offset += amt;
14301430 return amt;
14311431 }
std/event/loop.zig+37-29
......@@ -98,9 +98,21 @@ pub const Loop = struct {
9898 };
9999 pub const instance: ?*Loop = if (@hasDecl(root, "event_loop")) root.event_loop else default_instance;
100100
101 /// TODO copy elision / named return values so that the threads referencing *Loop
102 /// have the correct pointer value.
103 /// https://github.com/ziglang/zig/issues/2761 and https://github.com/ziglang/zig/issues/2765
104 pub fn init(self: *Loop, allocator: *mem.Allocator) !void {
105 if (builtin.single_threaded) {
106 return self.initSingleThreaded(allocator);
107 } else {
108 return self.initMultiThreaded(allocator);
109 }
110 }
111
101112 /// After initialization, call run().
102113 /// TODO copy elision / named return values so that the threads referencing *Loop
103114 /// have the correct pointer value.
115 /// https://github.com/ziglang/zig/issues/2761 and https://github.com/ziglang/zig/issues/2765
104116 pub fn initSingleThreaded(self: *Loop, allocator: *mem.Allocator) !void {
105117 return self.initInternal(allocator, 1);
106118 }
......@@ -110,6 +122,7 @@ pub const Loop = struct {
110122 /// After initialization, call run().
111123 /// TODO copy elision / named return values so that the threads referencing *Loop
112124 /// have the correct pointer value.
125 /// https://github.com/ziglang/zig/issues/2761 and https://github.com/ziglang/zig/issues/2765
113126 pub fn initMultiThreaded(self: *Loop, allocator: *mem.Allocator) !void {
114127 if (builtin.single_threaded) @compileError("initMultiThreaded unavailable when building in single-threaded mode");
115128 const core_count = try Thread.cpuCount();
......@@ -161,18 +174,18 @@ pub const Loop = struct {
161174 fn initOsData(self: *Loop, extra_thread_count: usize) InitOsDataError!void {
162175 switch (builtin.os) {
163176 .linux => {
164 // TODO self.os_data.fs_queue = std.atomic.Queue(fs.Request).init();
165 // TODO self.os_data.fs_queue_item = 0;
166 // TODO // we need another thread for the file system because Linux does not have an async
167 // TODO // file system I/O API.
168 // TODO self.os_data.fs_end_request = fs.RequestNode{
169 // TODO .prev = undefined,
170 // TODO .next = undefined,
171 // TODO .data = fs.Request{
172 // TODO .msg = fs.Request.Msg.End,
173 // TODO .finish = fs.Request.Finish.NoAction,
174 // TODO },
175 // TODO };
177 self.os_data.fs_queue = std.atomic.Queue(fs.Request).init();
178 self.os_data.fs_queue_item = 0;
179 // we need another thread for the file system because Linux does not have an async
180 // file system I/O API.
181 self.os_data.fs_end_request = fs.RequestNode{
182 .prev = undefined,
183 .next = undefined,
184 .data = fs.Request{
185 .msg = fs.Request.Msg.End,
186 .finish = fs.Request.Finish.NoAction,
187 },
188 };
176189
177190 errdefer {
178191 while (self.available_eventfd_resume_nodes.pop()) |node| os.close(node.data.eventfd);
......@@ -210,10 +223,10 @@ pub const Loop = struct {
210223 &self.os_data.final_eventfd_event,
211224 );
212225
213 // TODO self.os_data.fs_thread = try Thread.spawn(self, posixFsRun);
226 self.os_data.fs_thread = try Thread.spawn(self, posixFsRun);
214227 errdefer {
215 // TODO self.posixFsRequest(&self.os_data.fs_end_request);
216 // TODO self.os_data.fs_thread.wait();
228 self.posixFsRequest(&self.os_data.fs_end_request);
229 self.os_data.fs_thread.wait();
217230 }
218231
219232 if (builtin.single_threaded) {
......@@ -315,10 +328,10 @@ pub const Loop = struct {
315328 .udata = undefined,
316329 };
317330
318 // TODO self.os_data.fs_thread = try Thread.spawn(self, posixFsRun);
331 self.os_data.fs_thread = try Thread.spawn(self, posixFsRun);
319332 errdefer {
320 // TODO self.posixFsRequest(&self.os_data.fs_end_request);
321 // TODO self.os_data.fs_thread.wait();
333 self.posixFsRequest(&self.os_data.fs_end_request);
334 self.os_data.fs_thread.wait();
322335 }
323336
324337 if (builtin.single_threaded) {
......@@ -441,7 +454,6 @@ pub const Loop = struct {
441454 pub async fn linuxWaitFd(self: *Loop, fd: i32, flags: u32) !void {
442455 defer self.linuxRemoveFd(fd);
443456 suspend {
444 // TODO explicitly put this memory in the coroutine frame #1194
445457 var resume_node = ResumeNode.Basic{
446458 .base = ResumeNode{
447459 .id = ResumeNode.Id.Basic,
......@@ -454,10 +466,6 @@ pub const Loop = struct {
454466 }
455467
456468 pub async fn bsdWaitKev(self: *Loop, ident: usize, filter: i16, fflags: u32) !os.Kevent {
457 // TODO #1194
458 suspend {
459 resume @handle();
460 }
461469 var resume_node = ResumeNode.Basic{
462470 .base = ResumeNode{
463471 .id = ResumeNode.Id.Basic,
......@@ -578,7 +586,7 @@ pub const Loop = struct {
578586 .macosx,
579587 .freebsd,
580588 .netbsd,
581 => {}, // TODO self.os_data.fs_thread.wait(),
589 => self.os_data.fs_thread.wait(),
582590 else => {},
583591 }
584592
......@@ -631,7 +639,7 @@ pub const Loop = struct {
631639 // cause all the threads to stop
632640 switch (builtin.os) {
633641 .linux => {
634 // TODO self.posixFsRequest(&self.os_data.fs_end_request);
642 self.posixFsRequest(&self.os_data.fs_end_request);
635643 // writing 8 bytes to an eventfd cannot fail
636644 os.write(self.os_data.final_eventfd, wakeup_bytes) catch unreachable;
637645 return;
......@@ -862,10 +870,10 @@ pub const Loop = struct {
862870 epollfd: i32,
863871 final_eventfd: i32,
864872 final_eventfd_event: os.linux.epoll_event,
865 // TODO fs_thread: *Thread,
866 // TODO fs_queue_item: i32,
867 // TODO fs_queue: std.atomic.Queue(fs.Request),
868 // TODO fs_end_request: fs.RequestNode,
873 fs_thread: *Thread,
874 fs_queue_item: i32,
875 fs_queue: std.atomic.Queue(fs.Request),
876 fs_end_request: fs.RequestNode,
869877 };
870878};
871879