diff --git a/src/datastruct/main.zig b/src/datastruct/main.zig index b3bdc6a99..73e9afba9 100644 --- a/src/datastruct/main.zig +++ b/src/datastruct/main.zig @@ -5,7 +5,6 @@ const blocking_queue = @import("blocking_queue.zig"); const cache_table = @import("cache_table.zig"); const circ_buf = @import("circ_buf.zig"); const intrusive_linked_list = @import("intrusive_linked_list.zig"); -const segmented_pool = @import("segmented_pool.zig"); const split_tree = @import("split_tree.zig"); pub const BlockingQueue = blocking_queue.BlockingQueue; @@ -14,7 +13,6 @@ pub const CircBuf = circ_buf.CircBuf; pub const IntrusiveDoublyLinkedList = intrusive_linked_list.DoublyLinkedList; pub const LimitedAllocator = @import("limited_allocator.zig").LimitedAllocator; pub const MessageData = @import("message_data.zig").MessageData; -pub const SegmentedPool = segmented_pool.SegmentedPool; pub const SplitTree = split_tree.SplitTree; test { diff --git a/src/datastruct/segmented_pool.zig b/src/datastruct/segmented_pool.zig deleted file mode 100644 index 4ffc5d59f..000000000 --- a/src/datastruct/segmented_pool.zig +++ /dev/null @@ -1,96 +0,0 @@ -const std = @import("std"); -const assert = @import("../quirks.zig").inlineAssert; -const Allocator = std.mem.Allocator; -const SegmentedList = @import("segmented_list.zig").SegmentedList; -const testing = std.testing; - -/// A data structure where you can get stable (never copied) pointers to -/// a type that automatically grows if necessary. The values can be "put back" -/// but are expected to be put back IN ORDER. -/// -/// This is implemented specifically for libuv write requests, since the -/// write requests must have a stable pointer and are guaranteed to be processed -/// in order for a single stream. -/// -/// This is NOT thread safe. -pub fn SegmentedPool(comptime T: type, comptime prealloc: usize) type { - return struct { - const Self = @This(); - - i: usize = 0, - available: usize = prealloc, - list: SegmentedList(T, prealloc) = .{ .len = prealloc }, - - pub fn deinit(self: *Self, alloc: Allocator) void { - self.list.deinit(alloc); - self.* = undefined; - } - - /// Get the next available value out of the list. This will not - /// grow the list. - pub fn get(self: *Self) !*T { - // Error to not have any - if (self.available == 0) return error.OutOfValues; - - // The index we grab is just i % len, so we wrap around to the front. - const i = @mod(self.i, self.list.len); - self.i +%= 1; // Wrapping addition to swe go back to 0 - self.available -= 1; - return self.list.at(i); - } - - /// Get the next available value out of the list and grow the list - /// if necessary. - pub fn getGrow(self: *Self, alloc: Allocator) !*T { - if (self.available == 0) try self.grow(alloc); - return try self.get(); - } - - fn grow(self: *Self, alloc: Allocator) !void { - try self.list.growCapacity(alloc, self.list.len * 2); - self.i = self.list.len; - self.available = self.list.len; - self.list.len *= 2; - } - - /// Put a value back. The value put back is expected to be the - /// in order of get. - pub fn put(self: *Self) void { - self.available += 1; - assert(self.available <= self.list.len); - } - }; -} - -test "SegmentedPool" { - var list: SegmentedPool(u8, 2) = .{}; - defer list.deinit(testing.allocator); - try testing.expectEqual(@as(usize, 2), list.available); - - // Get to capacity - const v1 = try list.get(); - const v2 = try list.get(); - try testing.expect(v1 != v2); - try testing.expectError(error.OutOfValues, list.get()); - - // Test writing for later - v1.* = 42; - - // Put a value back - list.put(); - const temp = try list.get(); - try testing.expect(v1 == temp); - try testing.expect(temp.* == 42); - try testing.expectError(error.OutOfValues, list.get()); - - // Grow - const v3 = try list.getGrow(testing.allocator); - try testing.expect(v1 != v3 and v2 != v3); - _ = try list.get(); - try testing.expectError(error.OutOfValues, list.get()); - - // Put a value back - list.put(); - try testing.expect(v1 == try list.get()); - try testing.expectError(error.OutOfValues, list.get()); -} diff --git a/src/termio/Exec.zig b/src/termio/Exec.zig index 7b520a759..016291712 100644 --- a/src/termio/Exec.zig +++ b/src/termio/Exec.zig @@ -22,7 +22,6 @@ const shell_integration = @import("shell_integration.zig"); const terminal = @import("../terminal/main.zig"); const termio = @import("../termio.zig"); const Command = @import("../Command.zig"); -const SegmentedPool = @import("../datastruct/main.zig").SegmentedPool; const ptypkg = @import("../pty.zig"); const Pty = ptypkg.Pty; const EnvMap = std.process.Environ.Map; @@ -418,8 +417,9 @@ pub fn queueWrite( // our cached buffers that we can queue to the stream. var i: usize = 0; while (i < data.len) { - const req = try exec.write_req_pool.getGrow(alloc); - const buf = try exec.write_buf_pool.getGrow(alloc); + const w = try exec.write_pool.create(alloc); + w.td = exec; + const buf = &w.buf; const slice = slice: { // The maximum end index is either the end of our data or // the end of our buffer, whichever is smaller. @@ -459,26 +459,25 @@ pub fn queueWrite( exec.write_stream.queueWrite( td.loop, &exec.write_queue, - req, + &w.req, .{ .slice = slice }, - termio.Exec.ThreadData, - exec, + ThreadData.Write, + w, ttyWrite, ); } } fn ttyWrite( - td_: ?*ThreadData, + w_: ?*ThreadData.Write, _: *xev.Loop, _: *xev.Completion, _: xev.Stream, _: xev.WriteBuffer, r: xev.WriteError!usize, ) xev.CallbackAction { - const td = td_.?; - td.write_req_pool.put(); - td.write_buf_pool.put(); + const w = w_.?; + w.td.write_pool.destroy(w); const d = r catch |err| { log.err("write error: {}", .{err}); @@ -492,9 +491,21 @@ fn ttyWrite( /// The thread local data for the exec implementation. pub const ThreadData = struct { - // The preallocation size for the write request pool. This should be big - // enough to satisfy most write requests. It must be a power of 2. - const WRITE_REQ_PREALLOC = std.math.pow(usize, 2, 5); + /// The state for a single queued pty write. The write request and + /// the buffer it writes from must both remain pointer-stable until + /// the write completes, so they're pooled together and checked out + /// per write. + pub const Write = struct { + /// Backpointer to the thread data so the write completion + /// callback can put this back into the pool. + td: *ThreadData, + + /// The libxev write request. + req: xev.WriteRequest, + + /// The buffer for the data being written. + buf: [64]u8, + }; /// Process start time and boolean of whether its already exited. start: std.Io.Timestamp, @@ -506,12 +517,9 @@ pub const ThreadData = struct { /// The process watcher process: ?xev.Process, - /// This is the pool of available (unused) write requests. If you grab + /// This is the pool of available (unused) write states. If you grab /// one from the pool, you must put it back when you're done! - write_req_pool: SegmentedPool(xev.WriteRequest, WRITE_REQ_PREALLOC) = .{}, - - /// The pool of available buffers for writing to the pty. - write_buf_pool: SegmentedPool([64]u8, WRITE_REQ_PREALLOC) = .{}, + write_pool: std.heap.MemoryPool(Write) = .empty, /// The write queue for the data stream. write_queue: xev.WriteQueue = .{}, @@ -541,11 +549,10 @@ pub const ThreadData = struct { pub fn deinit(self: *ThreadData, alloc: Allocator) void { _ = posix.system.close(self.read_thread_pipe); - // Clear our write pools. We know we aren't ever going to do + // Clear our write pool. We know we aren't ever going to do // any more IO since we stop our data stream below so we can just // drop this. - self.write_req_pool.deinit(alloc); - self.write_buf_pool.deinit(alloc); + self.write_pool.deinit(alloc); // Stop our process watcher if (self.process) |*p| p.deinit();