pipe selections through shell commands

1 files changed, 59 insertions, 2 deletions

diff --git a/src/tty/tty.zig b/src/tty/tty.zig
index 90477d06..908f17d7 100644
--- a/src/tty/tty.zig
+++ b/src/tty/tty.zig
@@ -12,6 +12,7 @@ const pardes = @import("../pardes.zig");
 const tracy = @import("../tracy.zig");
 const look = @import("../look.zig");
 const temp_file = @import("../temp_file.zig");
+const selection_pipe = @import("../selection_pipe.zig");
 
 extern "c" fn forkpty(amaster: *c_int, name: ?[*:0]u8, termp: ?*const anyopaque, winp: ?*const posix.winsize) c_int;
 extern "c" fn execv(path: [*:0]const u8, argv: [*:null]const ?[*:0]const u8) c_int;
@@ -38,6 +39,8 @@ pub const Command = struct {
         paste: []const u8,
         /// a language query finished on a worker; rows are gpa-owned
         lsp_done: struct { id: u32, rows: []u8 },
+        /// a selection-filter worker finished; every stdout is gpa-owned
+        pipe_done: selection_pipe.Response,
         /// something happened in a watched directory (see watchFiles)
         files_changed,
     } = .nop;
@@ -86,6 +89,11 @@ const LspJob = struct {
     }
 };
 
+const PipeTask = struct {
+    id: u32,
+    future: std.Io.Future(anyerror!void),
+};
+
 const Pty = struct {
     file: std.Io.File,
     pid: posix.pid_t,
@@ -387,6 +395,9 @@ pub fn run(init: std.process.Init, opts: pardes.Options) !void {
     var gens: [pardes.MAX_PANES]u32 = @splat(0);
     // the single in-flight language query (see the .lsp effect)
     var lsp_task: ?std.Io.Future(anyerror!void) = null;
+    // Selection filters may overlap: a second submit supersedes the first in
+    // core without synchronously canceling a possibly slow shell command.
+    var pipe_tasks: std.ArrayList(PipeTask) = .empty;
     // One inotify instance for every watched pane, opened here — before any
     // thread exists — so the pre-loop effect drain below can already mark the
     // file a positional path argument opened. -1 off linux: watchPane goes
@@ -411,6 +422,8 @@ pub fn run(init: std.process.Init, opts: pardes.Options) !void {
             t.cancel(io) catch {};
             lsp_task = null;
         }
+        for (pipe_tasks.items) |*task| task.future.cancel(io) catch {};
+        pipe_tasks.deinit(gpa);
         // same contract as the pty readers: cancel unblocks the watcher's
         // read, and only then is the fd safe to close
         if (watch_task) |*t| {
@@ -425,13 +438,17 @@ pub fn run(init: std.process.Init, opts: pardes.Options) !void {
             .pty_read => |pr| gpa.free(pr.bytes),
             .paste => |b| gpa.free(@constCast(b)),
             .lsp_done => |d| gpa.free(d.rows),
+            .pipe_done => |response_value| {
+                var response = response_value;
+                response.deinit(gpa);
+            },
             else => {},
         };
     }
 
     // Perform the initial spawns BEFORE any worker thread exists: forkpty from
     // a multithreaded process can wedge the child before exec.
-    drainEffects(core, &ptys, &gens, io, gpa, &loop, &vx, &tty, &lsp_task, inotify_fd, &watches, false);
+    drainEffects(core, &ptys, &gens, io, gpa, &loop, &vx, &tty, &lsp_task, &pipe_tasks, inotify_fd, &watches, false);
 
     try loop.start();
     defer loop.stop();
@@ -589,6 +606,20 @@ pub fn run(init: std.process.Init, opts: pardes.Options) !void {
                         lsp_task = null;
                     }
                 },
+                .pipe_done => |response_value| {
+                    var response = response_value;
+                    core.update(.{ .pipe_resp = .{
+                        .id = response.id,
+                        .success = response.success,
+                        .outputs = response.outputs,
+                    } });
+                    response.deinit(gpa);
+                    for (pipe_tasks.items, 0..) |*task, i| if (task.id == response_value.id) {
+                        task.future.await(io) catch {};
+                        _ = pipe_tasks.orderedRemove(i);
+                        break;
+                    };
+                },
             }
             batch += 1;
             if (stop or output or native_pdf_page_changed or batch >= 64) break;
@@ -619,7 +650,7 @@ pub fn run(init: std.process.Init, opts: pardes.Options) !void {
             }
         }
 
-        drainEffects(core, &ptys, &gens, io, gpa, &loop, &vx, &tty, &lsp_task, inotify_fd, &watches, true);
+        drainEffects(core, &ptys, &gens, io, gpa, &loop, &vx, &tty, &lsp_task, &pipe_tasks, inotify_fd, &watches, true);
         if (core.quit) break;
 
         // Restore builtin: swap in a core rebuilt from the dump; the live
@@ -781,6 +812,7 @@ fn drainEffects(
     vx: *vaxis.Vaxis,
     tty: *vaxis.Tty,
     lsp_task: *?std.Io.Future(anyerror!void),
+    pipe_tasks: *std.ArrayList(PipeTask),
     inotify_fd: c_int,
     watches: *[pardes.MAX_PANES]?Watch,
     threads_ok: bool,
@@ -910,6 +942,20 @@ fn drainEffects(
                 continue;
             };
         },
+        .pipe => |request| {
+            if (!threads_ok) continue;
+            const view = core.pipeRequest(request.id) orelse continue;
+            const job = selection_pipe.Job.copy(gpa, view) catch continue;
+            pipe_tasks.ensureUnusedCapacity(gpa, 1) catch {
+                job.deinit(gpa);
+                continue;
+            };
+            const future = io.concurrent(pipeWorker, .{ io, gpa, job, loop }) catch {
+                job.deinit(gpa);
+                continue;
+            };
+            pipe_tasks.appendAssumeCapacity(.{ .id = request.id, .future = future });
+        },
         .watch => |w| {
             // starting, the path and the on-disk bytes are read off the core
             // (same split as save_file); stopping, the pane is already gone
@@ -949,6 +995,17 @@ fn lspWorker(gpa: std.mem.Allocator, job: *LspJob, loop: *Loop) anyerror!void {
     return;
 }
 
+fn pipeWorker(
+    io: std.Io,
+    gpa: std.mem.Allocator,
+    job: *selection_pipe.Job,
+    loop: *Loop,
+) anyerror!void {
+    defer job.deinit(gpa);
+    var response = selection_pipe.runJob(gpa, io, job);
+    loop.postEvent(.{ .pipe_done = response }) catch response.deinit(gpa);
+}
+
 /// Mark or unmark one pane's file (`path` null = unmark). Linux only: on any
 /// other OS this returns silently, the core never receives a file_changed
 /// event, and the feature is simply off — a state the core has to tolerate