neurocyte/flow
neurocyte/flow
1
1
Fork 0
Code Issues Pull requests Projects Releases 5 Packages Wiki Activity Actions

refactor: drop watcher Backend.threaded

Browse source
This commit is contained in:
CJ van den Berg 2026-02-20 19:36:16 +01:00
parent 3d90fb724b
commit aa5f189175
Signed by: neurocyte
GPG key ID: 8EB1E1BB660E3FB9
1 changed files with 7 additions and 13 deletions
Download patch file Download diff file Expand all files Collapse all files

20
src/file_watcher.zig
View file

@ -78,8 +78,6 @@ const INotifyBackend = struct {
fd_watcher: tp.file_descriptor, fd_watcher: tp.file_descriptor,
watches: std.AutoHashMapUnmanaged(i32, []u8), // wd -> owned path watches: std.AutoHashMapUnmanaged(i32, []u8), // wd -> owned path
const threaded = false;
const IN = std.os.linux.IN; const IN = std.os.linux.IN;
const watch_mask: u32 = IN.CREATE | IN.DELETE | IN.MODIFY | const watch_mask: u32 = IN.CREATE | IN.DELETE | IN.MODIFY |
@ -131,7 +129,7 @@ const INotifyBackend = struct {
} }
} }
fn drain(self: *@This(), allocator: std.mem.Allocator, parent: tp.pid_ref) (std.posix.ReadError || error{ NoSpaceLeft, OutOfMemory, Exit })!void { fn handle_read_ready(self: *@This(), allocator: std.mem.Allocator, parent: tp.pid_ref) (std.posix.ReadError || error{ NoSpaceLeft, OutOfMemory, Exit })!void {
const InotifyEvent = extern struct { const InotifyEvent = extern struct {
wd: i32, wd: i32,
mask: u32, mask: u32,
@ -228,7 +226,6 @@ const FSEventsBackend = struct {
watches: std.StringArrayHashMapUnmanaged(void), // owned paths watches: std.StringArrayHashMapUnmanaged(void), // owned paths
mutex: std.Thread.Mutex, // protects run_loop mutex: std.Thread.Mutex, // protects run_loop
const threaded = false; // callback sends FW messages directly; no FW_event needed
const kFSEventStreamCreateFlagNoDefer: u32 = 0x00000002; const kFSEventStreamCreateFlagNoDefer: u32 = 0x00000002;
const kFSEventStreamCreateFlagFileEvents: u32 = 0x00000010; const kFSEventStreamCreateFlagFileEvents: u32 = 0x00000010;
const kFSEventStreamEventFlagItemCreated: u32 = 0x00000100; const kFSEventStreamEventFlagItemCreated: u32 = 0x00000100;
@ -417,8 +414,6 @@ const KQueueBackend = struct {
thread: ?std.Thread, thread: ?std.Thread,
watches: std.StringHashMapUnmanaged(std.posix.fd_t), // owned path -> fd watches: std.StringHashMapUnmanaged(std.posix.fd_t), // owned path -> fd
const threaded = false; // events processed directly in thread_fn; no FW_event needed
const EVFILT_VNODE: i16 = -4; const EVFILT_VNODE: i16 = -4;
const EVFILT_READ: i16 = -1; const EVFILT_READ: i16 = -1;
const EV_ADD: u16 = 0x0001; const EV_ADD: u16 = 0x0001;
@ -528,7 +523,6 @@ const KQueueBackend = struct {
}; };
const WindowsBackend = struct { const WindowsBackend = struct {
const threaded = true;
const windows = std.os.windows; const windows = std.os.windows;
const win32 = struct { const win32 = struct {
@ -686,7 +680,7 @@ const WindowsBackend = struct {
} }
} }
fn drain(self: *@This(), allocator: std.mem.Allocator, parent: tp.pid_ref) !void { fn handle_fw_event(self: *@This(), allocator: std.mem.Allocator, parent: tp.pid_ref) !void {
_ = allocator; _ = allocator;
var bytes: windows.DWORD = 0; var bytes: windows.DWORD = 0;
var key: windows.ULONG_PTR = 0; var key: windows.ULONG_PTR = 0;
@ -790,12 +784,12 @@ const Process = struct {
var err_code: i64 = 0; var err_code: i64 = 0;
var err_msg: []const u8 = undefined; var err_msg: []const u8 = undefined;
if (!Backend.threaded and try cbor.match(m.buf, .{ "fd", tp.extract(&tag), "read_ready" })) { if (@hasDecl(Backend, "handle_read_ready") and try cbor.match(m.buf, .{ "fd", tp.extract(&tag), "read_ready" })) {
self.backend.drain(self.allocator, self.parent.ref()) catch |e| self.logger.err("drain", e); self.backend.handle_read_ready(self.allocator, self.parent.ref()) catch |e| self.logger.err("handle_read_ready", e);
} else if (!Backend.threaded and try cbor.match(m.buf, .{ "fd", tp.extract(&tag), "read_error", tp.extract(&err_code), tp.extract(&err_msg) })) { } else if (@hasDecl(Backend, "handle_read_ready") and try cbor.match(m.buf, .{ "fd", tp.extract(&tag), "read_error", tp.extract(&err_code), tp.extract(&err_msg) })) {
self.logger.print("fd read error on {s}: ({d}) {s}", .{ tag, err_code, err_msg }); self.logger.print("fd read error on {s}: ({d}) {s}", .{ tag, err_code, err_msg });
} else if (Backend.threaded and try cbor.match(m.buf, .{"FW_event"})) { } else if (@hasDecl(Backend, "handle_fw_event") and try cbor.match(m.buf, .{"FW_event"})) {
self.backend.drain(self.allocator, self.parent.ref()) catch |e| self.logger.err("drain", e); self.backend.handle_fw_event(self.allocator, self.parent.ref()) catch |e| self.logger.err("drain", e);
} else if (try cbor.match(m.buf, .{ "watch", tp.extract(&path) })) { } else if (try cbor.match(m.buf, .{ "watch", tp.extract(&path) })) {
self.backend.add_watch(self.allocator, path) catch |e| self.logger.err("watch", e); self.backend.add_watch(self.allocator, path) catch |e| self.logger.err("watch", e);
} else if (try cbor.match(m.buf, .{ "unwatch", tp.extract(&path) })) { } else if (try cbor.match(m.buf, .{ "unwatch", tp.extract(&path) })) {