| ... | ... | @@ -5,6 +5,8 @@ const assert = std.debug.assert; |
| 5 | 5 | const os = std.os; |
| 6 | 6 | const mem = std.mem; |
| 7 | 7 | const posix = os.posix; |
| 8 | const windows = os.windows; |
| 9 | const Loop = event.Loop; |
| 8 | 10 | |
| 9 | 11 | pub const RequestNode = std.atomic.Queue(Request).Node; |
| 10 | 12 | |
| ... | ... | @@ -13,7 +15,7 @@ pub const Request = struct { |
| 13 | 15 | finish: Finish, |
| 14 | 16 | |
| 15 | 17 | pub const Finish = union(enum) { |
| 16 | | TickNode: event.Loop.NextTickNode, |
| 18 | TickNode: Loop.NextTickNode, |
| 17 | 19 | DeallocCloseOperation: *CloseOperation, |
| 18 | 20 | NoAction, |
| 19 | 21 | }; |
| ... | ... | @@ -71,7 +73,77 @@ pub const Request = struct { |
| 71 | 73 | }; |
| 72 | 74 | |
| 73 | 75 | /// data - just the inner references - must live until pwritev promise completes. |
| 74 | | pub async fn pwritev(loop: *event.Loop, fd: os.FileHandle, data: []const []const u8, offset: usize) !void { |
| 76 | pub async fn pwritev(loop: *Loop, fd: os.FileHandle, data: []const []const u8, offset: usize) !void { |
| 77 | switch (builtin.os) { |
| 78 | builtin.Os.macosx, |
| 79 | builtin.Os.linux, |
| 80 | => return await (async pwritevPosix(loop, fd, data, offset) catch unreachable), |
| 81 | builtin.Os.windows, |
| 82 | => return await (async pwritevWindows(loop, fd, data, offset) catch unreachable), |
| 83 | else => @compileError("Unsupported OS"), |
| 84 | } |
| 85 | } |
| 86 | |
| 87 | /// data - just the inner references - must live until pwritev promise completes. |
| 88 | pub async fn pwritevWindows(loop: *Loop, fd: os.FileHandle, data: []const []const u8, offset: usize) !void { |
| 89 | if (data.len == 0) return; |
| 90 | if (data.len == 1) return await (async pwriteWindows(loop, fd, data[0], offset) catch unreachable); |
| 91 | |
| 92 | const data_copy = std.mem.dupe(loop.allocator, []const u8, data); |
| 93 | defer loop.allocator.free(data_copy); |
| 94 | |
| 95 | var off = offset; |
| 96 | for (data_copy) |buf| { |
| 97 | try await (async pwriteWindows(loop, fd, buf, off) catch unreachable); |
| 98 | off += buf.len; |
| 99 | } |
| 100 | } |
| 101 | |
| 102 | pub async fn pwriteWindows(loop: *Loop, fd: os.FileHandle, data: []const u8, offset: u64) os.WindowsWriteError!void { |
| 103 | // workaround for https://github.com/ziglang/zig/issues/1194 |
| 104 | suspend { |
| 105 | resume @handle(); |
| 106 | } |
| 107 | |
| 108 | var resume_node = Loop.ResumeNode.Basic{ |
| 109 | .base = Loop.ResumeNode{ |
| 110 | .id = Loop.ResumeNode.Id.Basic, |
| 111 | .handle = @handle(), |
| 112 | }, |
| 113 | }; |
| 114 | const completion_key = @ptrToInt(&resume_node.base); |
| 115 | _ = try os.windowsCreateIoCompletionPort(fd, loop.os_data.io_port, completion_key, undefined); |
| 116 | var overlapped = windows.OVERLAPPED{ |
| 117 | .Internal = 0, |
| 118 | .InternalHigh = 0, |
| 119 | .Offset = @truncate(u32, offset), |
| 120 | .OffsetHigh = @truncate(u32, offset >> 32), |
| 121 | .hEvent = null, |
| 122 | }; |
| 123 | errdefer { |
| 124 | _ = windows.CancelIoEx(fd, &overlapped); |
| 125 | } |
| 126 | suspend { |
| 127 | _ = windows.WriteFile(fd, data.ptr, @intCast(windows.DWORD, data.len), null, &overlapped); |
| 128 | } |
| 129 | var bytes_transferred: windows.DWORD = undefined; |
| 130 | if (windows.GetOverlappedResult(fd, &overlapped, &bytes_transferred, windows.FALSE) == 0) { |
| 131 | const err = windows.GetLastError(); |
| 132 | return switch (err) { |
| 133 | windows.ERROR.IO_PENDING => unreachable, |
| 134 | windows.ERROR.INVALID_USER_BUFFER => error.SystemResources, |
| 135 | windows.ERROR.NOT_ENOUGH_MEMORY => error.SystemResources, |
| 136 | windows.ERROR.OPERATION_ABORTED => error.OperationAborted, |
| 137 | windows.ERROR.NOT_ENOUGH_QUOTA => error.SystemResources, |
| 138 | windows.ERROR.BROKEN_PIPE => error.BrokenPipe, |
| 139 | else => os.unexpectedErrorWindows(err), |
| 140 | }; |
| 141 | } |
| 142 | } |
| 143 | |
| 144 | |
| 145 | /// data - just the inner references - must live until pwritev promise completes. |
| 146 | pub async fn pwritevPosix(loop: *Loop, fd: os.FileHandle, data: []const []const u8, offset: usize) !void { |
| 75 | 147 | // workaround for https://github.com/ziglang/zig/issues/1194 |
| 76 | 148 | suspend { |
| 77 | 149 | resume @handle(); |
| ... | ... | @@ -100,7 +172,7 @@ pub async fn pwritev(loop: *event.Loop, fd: os.FileHandle, data: []const []const |
| 100 | 172 | }, |
| 101 | 173 | }, |
| 102 | 174 | .finish = Request.Finish{ |
| 103 | | .TickNode = event.Loop.NextTickNode{ |
| 175 | .TickNode = Loop.NextTickNode{ |
| 104 | 176 | .prev = null, |
| 105 | 177 | .next = null, |
| 106 | 178 | .data = @handle(), |
| ... | ... | @@ -118,8 +190,8 @@ pub async fn pwritev(loop: *event.Loop, fd: os.FileHandle, data: []const []const |
| 118 | 190 | return req_node.data.msg.PWriteV.result; |
| 119 | 191 | } |
| 120 | 192 | |
| 121 | | /// data - just the inner references - must live until pwritev promise completes. |
| 122 | | pub async fn preadv(loop: *event.Loop, fd: os.FileHandle, data: []const []u8, offset: usize) !usize { |
| 193 | /// data - just the inner references - must live until preadv promise completes. |
| 194 | pub async fn preadv(loop: *Loop, fd: os.FileHandle, data: []const []u8, offset: usize) !usize { |
| 123 | 195 | //const data_dupe = try mem.dupe(loop.allocator, []const u8, data); |
| 124 | 196 | //defer loop.allocator.free(data_dupe); |
| 125 | 197 | |
| ... | ... | @@ -151,7 +223,7 @@ pub async fn preadv(loop: *event.Loop, fd: os.FileHandle, data: []const []u8, of |
| 151 | 223 | }, |
| 152 | 224 | }, |
| 153 | 225 | .finish = Request.Finish{ |
| 154 | | .TickNode = event.Loop.NextTickNode{ |
| 226 | .TickNode = Loop.NextTickNode{ |
| 155 | 227 | .prev = null, |
| 156 | 228 | .next = null, |
| 157 | 229 | .data = @handle(), |
| ... | ... | @@ -169,8 +241,8 @@ pub async fn preadv(loop: *event.Loop, fd: os.FileHandle, data: []const []u8, of |
| 169 | 241 | return req_node.data.msg.PReadV.result; |
| 170 | 242 | } |
| 171 | 243 | |
| 172 | | pub async fn open( |
| 173 | | loop: *event.Loop, |
| 244 | pub async fn openPosix( |
| 245 | loop: *Loop, |
| 174 | 246 | path: []const u8, |
| 175 | 247 | flags: u32, |
| 176 | 248 | mode: os.File.Mode, |
| ... | ... | @@ -196,7 +268,7 @@ pub async fn open( |
| 196 | 268 | }, |
| 197 | 269 | }, |
| 198 | 270 | .finish = Request.Finish{ |
| 199 | | .TickNode = event.Loop.NextTickNode{ |
| 271 | .TickNode = Loop.NextTickNode{ |
| 200 | 272 | .prev = null, |
| 201 | 273 | .next = null, |
| 202 | 274 | .data = @handle(), |
| ... | ... | @@ -214,19 +286,47 @@ pub async fn open( |
| 214 | 286 | return req_node.data.msg.Open.result; |
| 215 | 287 | } |
| 216 | 288 | |
| 217 | | pub async fn openRead(loop: *event.Loop, path: []const u8) os.File.OpenError!os.FileHandle { |
| 289 | pub async fn openRead(loop: *Loop, path: []const u8) os.File.OpenError!os.FileHandle { |
| 218 | 290 | const flags = posix.O_LARGEFILE | posix.O_RDONLY | posix.O_CLOEXEC; |
| 219 | | return await (async open(loop, path, flags, 0) catch unreachable); |
| 291 | return await (async openPosix(loop, path, flags, os.File.default_mode) catch unreachable); |
| 292 | } |
| 293 | |
| 294 | /// Creates if does not exist. Truncates the file if it exists. |
| 295 | /// Uses the default mode. |
| 296 | pub async fn openWrite(loop: *Loop, path: []const u8) os.File.OpenError!os.FileHandle { |
| 297 | return await (async openWriteMode(loop, path, os.File.default_mode) catch unreachable); |
| 298 | } |
| 299 | |
| 300 | /// Creates if does not exist. Truncates the file if it exists. |
| 301 | pub async fn openWriteMode(loop: *Loop, path: []const u8, mode: os.File.Mode) os.File.OpenError!os.FileHandle { |
| 302 | switch (builtin.os) { |
| 303 | builtin.Os.macosx, |
| 304 | builtin.Os.linux, |
| 305 | => { |
| 306 | const flags = posix.O_LARGEFILE | posix.O_WRONLY | posix.O_CREAT | posix.O_CLOEXEC | posix.O_TRUNC; |
| 307 | return await (async openPosix(loop, path, flags, os.File.default_mode) catch unreachable); |
| 308 | }, |
| 309 | builtin.Os.windows, |
| 310 | => return os.windowsOpen( |
| 311 | loop.allocator, |
| 312 | path, |
| 313 | windows.GENERIC_WRITE, |
| 314 | windows.FILE_SHARE_WRITE | windows.FILE_SHARE_READ | windows.FILE_SHARE_DELETE, |
| 315 | windows.CREATE_ALWAYS, |
| 316 | windows.FILE_ATTRIBUTE_NORMAL | windows.FILE_FLAG_OVERLAPPED, |
| 317 | ), |
| 318 | else => @compileError("Unsupported OS"), |
| 319 | } |
| 220 | 320 | } |
| 221 | 321 | |
| 222 | 322 | /// Creates if does not exist. Does not truncate. |
| 223 | 323 | pub async fn openReadWrite( |
| 224 | | loop: *event.Loop, |
| 324 | loop: *Loop, |
| 225 | 325 | path: []const u8, |
| 226 | 326 | mode: os.File.Mode, |
| 227 | 327 | ) os.File.OpenError!os.FileHandle { |
| 228 | 328 | const flags = posix.O_LARGEFILE | posix.O_RDWR | posix.O_CREAT | posix.O_CLOEXEC; |
| 229 | | return await (async open(loop, path, flags, mode) catch unreachable); |
| 329 | return await (async openPosix(loop, path, flags, mode) catch unreachable); |
| 230 | 330 | } |
| 231 | 331 | |
| 232 | 332 | /// This abstraction helps to close file handles in defer expressions |
| ... | ... | @@ -236,24 +336,46 @@ pub async fn openReadWrite( |
| 236 | 336 | /// If you call `setHandle` then finishing will close the fd; otherwise finishing |
| 237 | 337 | /// will deallocate the `CloseOperation`. |
| 238 | 338 | pub const CloseOperation = struct { |
| 239 | | loop: *event.Loop, |
| 240 | | have_fd: bool, |
| 241 | | close_req_node: RequestNode, |
| 339 | loop: *Loop, |
| 340 | os_data: OsData, |
| 341 | |
| 342 | const OsData = switch (builtin.os) { |
| 343 | builtin.Os.linux, |
| 344 | builtin.Os.macosx, |
| 345 | => struct { |
| 346 | have_fd: bool, |
| 347 | close_req_node: RequestNode, |
| 348 | }, |
| 349 | builtin.Os.windows, |
| 350 | => struct { |
| 351 | handle: ?os.FileHandle, |
| 352 | }, |
| 353 | else => @compileError("Unsupported OS"), |
| 354 | }; |
| 242 | 355 | |
| 243 | | pub fn start(loop: *event.Loop) (error{OutOfMemory}!*CloseOperation) { |
| 356 | pub fn start(loop: *Loop) (error{OutOfMemory}!*CloseOperation) { |
| 244 | 357 | const self = try loop.allocator.createOne(CloseOperation); |
| 245 | 358 | self.* = CloseOperation{ |
| 246 | 359 | .loop = loop, |
| 247 | | .have_fd = false, |
| 248 | | .close_req_node = RequestNode{ |
| 249 | | .prev = null, |
| 250 | | .next = null, |
| 251 | | .data = Request{ |
| 252 | | .msg = Request.Msg{ |
| 253 | | .Close = Request.Msg.Close{ .fd = undefined }, |
| 360 | .os_data = switch (builtin.os) { |
| 361 | builtin.Os.linux, |
| 362 | builtin.Os.macosx, |
| 363 | => OsData{ |
| 364 | .have_fd = false, |
| 365 | .close_req_node = RequestNode{ |
| 366 | .prev = null, |
| 367 | .next = null, |
| 368 | .data = Request{ |
| 369 | .msg = Request.Msg{ |
| 370 | .Close = Request.Msg.Close{ .fd = undefined }, |
| 371 | }, |
| 372 | .finish = Request.Finish{ .DeallocCloseOperation = self }, |
| 373 | }, |
| 254 | 374 | }, |
| 255 | | .finish = Request.Finish{ .DeallocCloseOperation = self }, |
| 256 | 375 | }, |
| 376 | builtin.Os.windows, |
| 377 | => OsData{ .handle = null }, |
| 378 | else => @compileError("Unsupported OS"), |
| 257 | 379 | }, |
| 258 | 380 | }; |
| 259 | 381 | return self; |
| ... | ... | @@ -261,36 +383,109 @@ pub const CloseOperation = struct { |
| 261 | 383 | |
| 262 | 384 | /// Defer this after creating. |
| 263 | 385 | pub fn finish(self: *CloseOperation) void { |
| 264 | | if (self.have_fd) { |
| 265 | | self.loop.posixFsRequest(&self.close_req_node); |
| 266 | | } else { |
| 267 | | self.loop.allocator.destroy(self); |
| 386 | switch (builtin.os) { |
| 387 | builtin.Os.linux, |
| 388 | builtin.Os.macosx, |
| 389 | => { |
| 390 | if (self.have_fd) { |
| 391 | self.loop.posixFsRequest(&self.close_req_node); |
| 392 | } else { |
| 393 | self.loop.allocator.destroy(self); |
| 394 | } |
| 395 | }, |
| 396 | builtin.Os.windows, |
| 397 | => { |
| 398 | if (self.handle) |handle| { |
| 399 | os.close(handle); |
| 400 | } |
| 401 | self.loop.allocator.destroy(self); |
| 402 | }, |
| 403 | else => @compileError("Unsupported OS"), |
| 268 | 404 | } |
| 269 | 405 | } |
| 270 | 406 | |
| 271 | 407 | pub fn setHandle(self: *CloseOperation, handle: os.FileHandle) void { |
| 272 | | self.close_req_node.data.msg.Close.fd = handle; |
| 273 | | self.have_fd = true; |
| 408 | switch (builtin.os) { |
| 409 | builtin.Os.linux, |
| 410 | builtin.Os.macosx, |
| 411 | => { |
| 412 | self.close_req_node.data.msg.Close.fd = handle; |
| 413 | self.have_fd = true; |
| 414 | }, |
| 415 | builtin.Os.windows, |
| 416 | => { |
| 417 | self.handle = handle; |
| 418 | }, |
| 419 | else => @compileError("Unsupported OS"), |
| 420 | } |
| 274 | 421 | } |
| 275 | 422 | |
| 276 | 423 | /// Undo a `setHandle`. |
| 277 | 424 | pub fn clearHandle(self: *CloseOperation) void { |
| 278 | | self.have_fd = false; |
| 425 | switch (builtin.os) { |
| 426 | builtin.Os.linux, |
| 427 | builtin.Os.macosx, |
| 428 | => { |
| 429 | self.have_fd = false; |
| 430 | }, |
| 431 | builtin.Os.windows, |
| 432 | => { |
| 433 | self.handle = null; |
| 434 | }, |
| 435 | else => @compileError("Unsupported OS"), |
| 436 | } |
| 279 | 437 | } |
| 280 | 438 | |
| 281 | 439 | pub fn getHandle(self: *CloseOperation) os.FileHandle { |
| 282 | | assert(self.have_fd); |
| 283 | | return self.close_req_node.data.msg.Close.fd; |
| 440 | switch (builtin.os) { |
| 441 | builtin.Os.linux, |
| 442 | builtin.Os.macosx, |
| 443 | => { |
| 444 | assert(self.have_fd); |
| 445 | return self.close_req_node.data.msg.Close.fd; |
| 446 | }, |
| 447 | builtin.Os.windows, |
| 448 | => { |
| 449 | return self.handle.?; |
| 450 | }, |
| 451 | else => @compileError("Unsupported OS"), |
| 452 | } |
| 284 | 453 | } |
| 285 | 454 | }; |
| 286 | 455 | |
| 287 | 456 | /// contents must remain alive until writeFile completes. |
| 288 | | pub async fn writeFile(loop: *event.Loop, path: []const u8, contents: []const u8) !void { |
| 457 | /// TODO make this atomic or provide writeFileAtomic and rename this one to writeFileTruncate |
| 458 | pub async fn writeFile(loop: *Loop, path: []const u8, contents: []const u8) !void { |
| 289 | 459 | return await (async writeFileMode(loop, path, contents, os.File.default_mode) catch unreachable); |
| 290 | 460 | } |
| 291 | 461 | |
| 292 | 462 | /// contents must remain alive until writeFile completes. |
| 293 | | pub async fn writeFileMode(loop: *event.Loop, path: []const u8, contents: []const u8, mode: os.File.Mode) !void { |
| 463 | pub async fn writeFileMode(loop: *Loop, path: []const u8, contents: []const u8, mode: os.File.Mode) !void { |
| 464 | switch (builtin.os) { |
| 465 | builtin.Os.linux, |
| 466 | builtin.Os.macosx, |
| 467 | => return await (async writeFileModeThread(loop, path, contents, mode) catch unreachable), |
| 468 | builtin.Os.windows, |
| 469 | => return await (async writeFileWindows(loop, path, contents) catch unreachable), |
| 470 | else => @compileError("Unsupported OS"), |
| 471 | } |
| 472 | } |
| 473 | |
| 474 | async fn writeFileWindows(loop: *Loop, path: []const u8, contents: []const u8) !void { |
| 475 | const handle = try os.windowsOpen( |
| 476 | loop.allocator, |
| 477 | path, |
| 478 | windows.GENERIC_WRITE, |
| 479 | windows.FILE_SHARE_WRITE | windows.FILE_SHARE_READ | windows.FILE_SHARE_DELETE, |
| 480 | windows.CREATE_ALWAYS, |
| 481 | windows.FILE_ATTRIBUTE_NORMAL | windows.FILE_FLAG_OVERLAPPED, |
| 482 | ); |
| 483 | defer os.close(handle); |
| 484 | |
| 485 | try await (async pwriteWindows(loop, handle, contents, 0) catch unreachable); |
| 486 | } |
| 487 | |
| 488 | async fn writeFileModeThread(loop: *Loop, path: []const u8, contents: []const u8, mode: os.File.Mode) !void { |
| 294 | 489 | // workaround for https://github.com/ziglang/zig/issues/1194 |
| 295 | 490 | suspend { |
| 296 | 491 | resume @handle(); |
| ... | ... | @@ -312,7 +507,7 @@ pub async fn writeFileMode(loop: *event.Loop, path: []const u8, contents: []cons |
| 312 | 507 | }, |
| 313 | 508 | }, |
| 314 | 509 | .finish = Request.Finish{ |
| 315 | | .TickNode = event.Loop.NextTickNode{ |
| 510 | .TickNode = Loop.NextTickNode{ |
| 316 | 511 | .prev = null, |
| 317 | 512 | .next = null, |
| 318 | 513 | .data = @handle(), |
| ... | ... | @@ -333,7 +528,7 @@ pub async fn writeFileMode(loop: *event.Loop, path: []const u8, contents: []cons |
| 333 | 528 | /// The promise resumes when the last data has been confirmed written, but before the file handle |
| 334 | 529 | /// is closed. |
| 335 | 530 | /// Caller owns returned memory. |
| 336 | | pub async fn readFile(loop: *event.Loop, file_path: []const u8, max_size: usize) ![]u8 { |
| 531 | pub async fn readFile(loop: *Loop, file_path: []const u8, max_size: usize) ![]u8 { |
| 337 | 532 | var close_op = try CloseOperation.start(loop); |
| 338 | 533 | defer close_op.finish(); |
| 339 | 534 | |
| ... | ... | @@ -417,7 +612,7 @@ pub fn Watch(comptime V: type) type { |
| 417 | 612 | pub const Error = WatchEventError; |
| 418 | 613 | }; |
| 419 | 614 | |
| 420 | | pub fn create(loop: *event.Loop, event_buf_count: usize) !*Self { |
| 615 | pub fn create(loop: *Loop, event_buf_count: usize) !*Self { |
| 421 | 616 | const channel = try event.Channel(Self.Event.Error!Self.Event).create(loop, event_buf_count); |
| 422 | 617 | errdefer channel.destroy(); |
| 423 | 618 | |
| ... | ... | @@ -482,7 +677,7 @@ pub fn Watch(comptime V: type) type { |
| 482 | 677 | |
| 483 | 678 | const flags = posix.O_SYMLINK | posix.O_EVTONLY; |
| 484 | 679 | const mode = 0; |
| 485 | | const fd = try await (async open(self.channel.loop, resolved_path, flags, mode) catch unreachable); |
| 680 | const fd = try await (async openPosix(self.channel.loop, resolved_path, flags, mode) catch unreachable); |
| 486 | 681 | close_op.setHandle(fd); |
| 487 | 682 | |
| 488 | 683 | var put_data: *OsData.Put = undefined; |
| ... | ... | @@ -722,7 +917,7 @@ test "write a file, watch it, write it again" { |
| 722 | 917 | try os.makePath(allocator, test_tmp_dir); |
| 723 | 918 | defer os.deleteTree(allocator, test_tmp_dir) catch {}; |
| 724 | 919 | |
| 725 | | var loop: event.Loop = undefined; |
| 920 | var loop: Loop = undefined; |
| 726 | 921 | try loop.initMultiThreaded(allocator); |
| 727 | 922 | defer loop.deinit(); |
| 728 | 923 | |
| ... | ... | @@ -734,11 +929,11 @@ test "write a file, watch it, write it again" { |
| 734 | 929 | return result; |
| 735 | 930 | } |
| 736 | 931 | |
| 737 | | async fn testFsWatchCantFail(loop: *event.Loop, result: *(error!void)) void { |
| 932 | async fn testFsWatchCantFail(loop: *Loop, result: *(error!void)) void { |
| 738 | 933 | result.* = await async testFsWatch(loop) catch unreachable; |
| 739 | 934 | } |
| 740 | 935 | |
| 741 | | async fn testFsWatch(loop: *event.Loop) !void { |
| 936 | async fn testFsWatch(loop: *Loop) !void { |
| 742 | 937 | const file_path = try os.path.join(loop.allocator, test_tmp_dir, "file.txt"); |
| 743 | 938 | defer loop.allocator.free(file_path); |
| 744 | 939 | |