Tweak cleanup code in PipeReader for files (#9746)

* WIP: some fixes and improvements * cleanup * WIP: some fixes and improvements * cleanup * dont pause --------- Co-authored-by: cirospaciari <ciro.spaciari@gmail.com>
oven-sh · Apr 1, 2024 · 5841721 · 5841721
1 parent 2d57f25
commit 5841721
Show file tree

Hide file tree

Showing 3 changed files with 27 additions and 21 deletions.
diff --git a/src/async/windows_event_loop.zig b/src/async/windows_event_loop.zig
@@ -398,27 +398,27 @@ pub const Closer = struct {
 
     pub fn close(fd: uv.uv_file, loop: *uv.Loop) void {
         var closer = Closer.new(.{});
-
-        if (uv.uv_fs_close(loop, &closer.io_request, fd, &onClose).errEnum()) |err| {
+        // data is not overridden by libuv when calling uv_fs_close, its ok to set it here
+        closer.io_request.data = closer;
+        if (uv.uv_fs_close(loop, &closer.io_request, fd, onClose).errEnum()) |err| {
             Output.debugWarn("libuv close() failed = {}", .{err});
             closer.destroy();
+            return;
         }
-
-        closer.io_request.data = closer;
     }
 
     fn onClose(req: *uv.fs_t) callconv(.C) void {
         var closer = @fieldParentPtr(Closer, "io_request", req);
         std.debug.assert(closer == @as(*Closer, @alignCast(@ptrCast(req.data.?))));
-        bun.sys.syslog("uv_fs_close() = {}", .{req.result});
+        bun.sys.syslog("uv_fs_close({}) = {}", .{ bun.toFD(req.file.fd), req.result });
 
         if (comptime Environment.allow_assert) {
             if (closer.io_request.result.errEnum()) |err| {
                 Output.debugWarn("libuv close() failed = {}", .{err});
             }
         }
 
-        uv.uv_fs_req_cleanup(req);
+        req.deinit();
         closer.destroy();
     }
 };
diff --git a/src/io/PipeReader.zig b/src/io/PipeReader.zig
@@ -472,9 +472,14 @@ pub fn WindowsPipeReader(
             if (this.source) |source| {
                 switch (source) {
                     .sync_file, .file => |file| {
+                        if (!this.flags.is_paused) {
+                            // always cancel the current one
+                            file.fs.cancel();
+                            this.flags.is_paused = true;
+                        }
                         // always use close_fs here because we can have a operation in progress
                         file.close_fs.data = file;
-                        _ = uv.uv_fs_close(uv.Loop.get(), &file.close_fs, file.file, @ptrCast(&onFileClose));
+                        _ = uv.uv_fs_close(uv.Loop.get(), &file.close_fs, file.file, onFileClose);
                     },
                     .pipe => |pipe| {
                         pipe.data = pipe;
@@ -532,10 +537,7 @@ pub fn WindowsPipeReader(
                 },
                 .drained => {
                     // we call drained so we know if we should stop here
-                    const keep_reading = onReadChunk(this, slice, hasMore);
-                    if (!keep_reading) {
-                        this.pause();
-                    }
+                    _ = onReadChunk(this, slice, hasMore);
                 },
                 else => {
                     var buffer = getBuffer(this);
@@ -546,10 +548,7 @@ pub fn WindowsPipeReader(
                     }
                     // move cursor foward
                     buffer.items.len += amount.result;
-                    const keep_reading = onReadChunk(this, slice, hasMore);
-                    if (!keep_reading) {
-                        this.pause();
-                    }
+                    _ = onReadChunk(this, slice, hasMore);
                 },
             }
         }

diff --git a/src/io/PipeWriter.zig b/src/io/PipeWriter.zig
@@ -785,13 +785,12 @@ fn BaseWindowsPipeWriter(
             this.is_done = true;
             if (this.source) |source| {
                 switch (source) {
-                    .file => |file| {
+                    .sync_file, .file => |file| {
+                        // always cancel the current one
+                        file.fs.cancel();
                         // always use close_fs here because we can have a operation in progress
                         file.close_fs.data = file;
-                        _ = uv.uv_fs_close(uv.Loop.get(), &file.close_fs, file.file, @ptrCast(&onFileClose));
-                    },
-                    .sync_file => {
-                        // no-op
+                        _ = uv.uv_fs_close(uv.Loop.get(), &file.close_fs, file.file, onFileClose);
                     },
                     .pipe => |pipe| {
                         pipe.data = pipe;
@@ -940,12 +939,20 @@ pub fn WindowsBufferedWriter(
         }
 
         fn onFsWriteComplete(fs: *uv.fs_t) callconv(.C) void {
+            const result = fs.result;
+            if (result.int() == uv.UV_ECANCELED) {
+                fs.deinit();
+                return;
+            }
             const this = bun.cast(*WindowsWriter, fs.data);
-            if (fs.result.toError(.write)) |err| {
+
+            fs.deinit();
+            if (result.toError(.write)) |err| {
                 this.close();
                 onError(this.parent, err);
                 return;
             }
+
             this.onWriteComplete(.zero);
         }