1//! A cross-platform interface that abstracts all I/O operations and
2//! concurrency. It includes:
3//! * file system
4//! * networking
5//! * processes
6//! * time and sleeping
7//! * randomness
8//! * async, await, concurrent, and cancel
9//! * concurrent queues
10//! * wait groups and select
11//! * mutexes, futexes, events, and conditions
12//! * memory mapped files
13//! This interface allows programmers to write optimal, reusable code while
14//! participating in these operations.
15const Io = @This();
16
17const builtin = @import("builtin");
18
19const std = @import("std.zig");
20const math = std.math;
21const assert = std.debug.assert;
22const Allocator = std.mem.Allocator;
23const Alignment = std.mem.Alignment;
24
25userdata: ?*anyopaque,
26vtable: *const VTable,
27
28pub const Threaded = @import("Io/Threaded.zig");
29
30pub const fiber = @import("Io/fiber.zig");
31pub const Evented = if (fiber.supported) switch (builtin.os.tag) {
32 .linux => Uring,
33 .dragonfly, .freebsd, .netbsd, .openbsd => Kqueue,
34 .driverkit, .ios, .maccatalyst, .macos, .tvos, .visionos, .watchos => Dispatch,
35 else => void,
36} else void; // context-switching code not implemented yet
37pub const Dispatch = @import("Io/Dispatch.zig");
38pub const Kqueue = @import("Io/Kqueue.zig");
39pub const Uring = @import("Io/Uring.zig");
40
41pub const Reader = @import("Io/Reader.zig");
42pub const Writer = @import("Io/Writer.zig");
43pub const net = @import("Io/net.zig");
44pub const Dir = @import("Io/Dir.zig");
45pub const File = @import("Io/File.zig");
46pub const Terminal = @import("Io/Terminal.zig");
47
48pub const RwLock = @import("Io/RwLock.zig");
49pub const Semaphore = @import("Io/Semaphore.zig");
50
51pub const VTable = struct {
52 crashHandler: *const fn (?*anyopaque) void,
53
54 /// If it returns `null` it means `result` has been already populated and
55 /// `await` will be a no-op.
56 ///
57 /// Thread-safe.
58 async: *const fn (
59 /// Corresponds to `Io.userdata`.
60 userdata: ?*anyopaque,
61 /// The pointer of this slice is an "eager" result value.
62 /// The length is the size in bytes of the result type.
63 /// This pointer's lifetime expires directly after the call to this function.
64 result: []u8,
65 result_alignment: std.mem.Alignment,
66 /// Copied and then passed to `start`.
67 context: []const u8,
68 context_alignment: std.mem.Alignment,
69 start: *const fn (context: *const anyopaque, result: *anyopaque) void,
70 ) ?*AnyFuture,
71 /// Thread-safe.
72 concurrent: *const fn (
73 /// Corresponds to `Io.userdata`.
74 userdata: ?*anyopaque,
75 result_len: usize,
76 result_alignment: std.mem.Alignment,
77 /// Copied and then passed to `start`.
78 context: []const u8,
79 context_alignment: std.mem.Alignment,
80 start: *const fn (context: *const anyopaque, result: *anyopaque) void,
81 ) ConcurrentError!*AnyFuture,
82 /// This function is only called when `async` returns a non-null value.
83 ///
84 /// Thread-safe.
85 await: *const fn (
86 /// Corresponds to `Io.userdata`.
87 userdata: ?*anyopaque,
88 /// The same value that was returned from `async`.
89 any_future: *AnyFuture,
90 /// Points to a buffer where the result is written.
91 /// The length is equal to size in bytes of result type.
92 result: []u8,
93 result_alignment: std.mem.Alignment,
94 ) void,
95 /// Equivalent to `await` but initiates cancel request.
96 ///
97 /// This function is only called when `async` returns a non-null value.
98 ///
99 /// Thread-safe.
100 cancel: *const fn (
101 /// Corresponds to `Io.userdata`.
102 userdata: ?*anyopaque,
103 /// The same value that was returned from `async`.
104 any_future: *AnyFuture,
105 /// Points to a buffer where the result is written.
106 /// The length is equal to size in bytes of result type.
107 result: []u8,
108 result_alignment: std.mem.Alignment,
109 ) void,
110
111 /// Thread-safe.
112 groupAsync: *const fn (
113 /// Corresponds to `Io.userdata`.
114 userdata: ?*anyopaque,
115 /// Owner of the spawned async task.
116 group: *Group,
117 /// Copied and then passed to `start`.
118 context: []const u8,
119 context_alignment: std.mem.Alignment,
120 start: *const fn (context: *const anyopaque) void,
121 ) void,
122 /// Thread-safe.
123 groupConcurrent: *const fn (
124 /// Corresponds to `Io.userdata`.
125 userdata: ?*anyopaque,
126 /// Owner of the spawned async task.
127 group: *Group,
128 /// Copied and then passed to `start`.
129 context: []const u8,
130 context_alignment: std.mem.Alignment,
131 start: *const fn (context: *const anyopaque) void,
132 ) ConcurrentError!void,
133 groupAwait: *const fn (?*anyopaque, *Group, token: *anyopaque) Cancelable!void,
134 groupCancel: *const fn (?*anyopaque, *Group, token: *anyopaque) void,
135
136 recancel: *const fn (?*anyopaque) void,
137 swapCancelProtection: *const fn (?*anyopaque, new: CancelProtection) CancelProtection,
138 checkCancel: *const fn (?*anyopaque) Cancelable!void,
139
140 futexWait: *const fn (?*anyopaque, ptr: *const u32, expected: u32, Timeout) Cancelable!void,
141 futexWaitUncancelable: *const fn (?*anyopaque, ptr: *const u32, expected: u32) void,
142 futexWake: *const fn (?*anyopaque, ptr: *const u32, max_waiters: u32) void,
143
144 operate: *const fn (?*anyopaque, Operation) Cancelable!Operation.Result,
145 batchAwaitAsync: *const fn (?*anyopaque, *Batch) Cancelable!void,
146 batchAwaitConcurrent: *const fn (?*anyopaque, *Batch, Timeout) Batch.AwaitConcurrentError!void,
147 batchCancel: *const fn (?*anyopaque, *Batch) void,
148
149 dirCreateDir: *const fn (?*anyopaque, Dir, []const u8, Dir.Permissions) Dir.CreateDirError!void,
150 dirCreateDirPath: *const fn (?*anyopaque, Dir, []const u8, Dir.Permissions) Dir.CreateDirPathError!Dir.CreatePathStatus,
151 dirCreateDirPathOpen: *const fn (?*anyopaque, Dir, []const u8, Dir.Permissions, Dir.OpenOptions) Dir.CreateDirPathOpenError!Dir,
152 dirOpenDir: *const fn (?*anyopaque, Dir, []const u8, Dir.OpenOptions) Dir.OpenError!Dir,
153 dirStat: *const fn (?*anyopaque, Dir) Dir.StatError!Dir.Stat,
154 dirStatFile: *const fn (?*anyopaque, Dir, []const u8, Dir.StatFileOptions) Dir.StatFileError!File.Stat,
155 dirAccess: *const fn (?*anyopaque, Dir, []const u8, Dir.AccessOptions) Dir.AccessError!void,
156 dirCreateFile: *const fn (?*anyopaque, Dir, []const u8, Dir.CreateFileOptions) File.OpenError!File,
157 dirCreateFileAtomic: *const fn (?*anyopaque, Dir, []const u8, Dir.CreateFileAtomicOptions) Dir.CreateFileAtomicError!File.Atomic,
158 dirOpenFile: *const fn (?*anyopaque, Dir, []const u8, Dir.OpenFileOptions) File.OpenError!File,
159 dirClose: *const fn (?*anyopaque, []const Dir) void,
160 dirRead: *const fn (?*anyopaque, *Dir.Reader, []Dir.Entry) Dir.Reader.Error!usize,
161 dirRealPath: *const fn (?*anyopaque, Dir, out_buffer: []u8) Dir.RealPathError!usize,
162 dirRealPathFile: *const fn (?*anyopaque, Dir, path_name: []const u8, out_buffer: []u8) Dir.RealPathFileError!usize,
163 dirDeleteFile: *const fn (?*anyopaque, Dir, []const u8) Dir.DeleteFileError!void,
164 dirDeleteDir: *const fn (?*anyopaque, Dir, []const u8) Dir.DeleteDirError!void,
165 dirRename: *const fn (?*anyopaque, old_dir: Dir, old_sub_path: []const u8, new_dir: Dir, new_sub_path: []const u8) Dir.RenameError!void,
166 dirRenamePreserve: *const fn (?*anyopaque, old_dir: Dir, old_sub_path: []const u8, new_dir: Dir, new_sub_path: []const u8) Dir.RenamePreserveError!void,
167 dirSymLink: *const fn (?*anyopaque, Dir, target_path: []const u8, sym_link_path: []const u8, Dir.SymLinkFlags) Dir.SymLinkError!void,
168 dirReadLink: *const fn (?*anyopaque, Dir, sub_path: []const u8, buffer: []u8) Dir.ReadLinkError!usize,
169 dirSetOwner: *const fn (?*anyopaque, Dir, ?File.Uid, ?File.Gid) Dir.SetOwnerError!void,
170 dirSetFileOwner: *const fn (?*anyopaque, Dir, []const u8, ?File.Uid, ?File.Gid, Dir.SetFileOwnerOptions) Dir.SetFileOwnerError!void,
171 dirSetPermissions: *const fn (?*anyopaque, Dir, Dir.Permissions) Dir.SetPermissionsError!void,
172 dirSetFilePermissions: *const fn (?*anyopaque, Dir, []const u8, File.Permissions, Dir.SetFilePermissionsOptions) Dir.SetFilePermissionsError!void,
173 dirSetTimestamps: *const fn (?*anyopaque, Dir, []const u8, Dir.SetTimestampsOptions) Dir.SetTimestampsError!void,
174 dirHardLink: *const fn (?*anyopaque, old_dir: Dir, old_sub_path: []const u8, new_dir: Dir, new_sub_path: []const u8, Dir.HardLinkOptions) Dir.HardLinkError!void,
175
176 fileStat: *const fn (?*anyopaque, File) File.StatError!File.Stat,
177 fileLength: *const fn (?*anyopaque, File) File.LengthError!u64,
178 fileClose: *const fn (?*anyopaque, []const File) void,
179 fileWritePositional: *const fn (?*anyopaque, File, header: []const u8, data: []const []const u8, splat: usize, offset: u64) File.WritePositionalError!usize,
180 fileWriteFileStreaming: *const fn (?*anyopaque, File, header: []const u8, *Io.File.Reader, Io.Limit) File.Writer.WriteFileError!usize,
181 fileWriteFilePositional: *const fn (?*anyopaque, File, header: []const u8, *Io.File.Reader, Io.Limit, offset: u64) File.WriteFilePositionalError!usize,
182 /// Returns 0 if reading at or past the end.
183 fileReadPositional: *const fn (?*anyopaque, File, data: []const []u8, offset: u64) File.ReadPositionalError!usize,
184 fileSeekBy: *const fn (?*anyopaque, File, relative_offset: i64) File.SeekError!void,
185 fileSeekTo: *const fn (?*anyopaque, File, absolute_offset: u64) File.SeekError!void,
186 fileSync: *const fn (?*anyopaque, File) File.SyncError!void,
187 fileIsTty: *const fn (?*anyopaque, File) Cancelable!bool,
188 fileEnableAnsiEscapeCodes: *const fn (?*anyopaque, File) File.EnableAnsiEscapeCodesError!void,
189 fileSupportsAnsiEscapeCodes: *const fn (?*anyopaque, File) Cancelable!bool,
190 fileSetLength: *const fn (?*anyopaque, File, u64) File.SetLengthError!void,
191 fileSetOwner: *const fn (?*anyopaque, File, ?File.Uid, ?File.Gid) File.SetOwnerError!void,
192 fileSetPermissions: *const fn (?*anyopaque, File, File.Permissions) File.SetPermissionsError!void,
193 fileSetTimestamps: *const fn (?*anyopaque, File, File.SetTimestampsOptions) File.SetTimestampsError!void,
194 fileLock: *const fn (?*anyopaque, File, File.Lock) File.LockError!void,
195 fileTryLock: *const fn (?*anyopaque, File, File.Lock) File.LockError!bool,
196 fileUnlock: *const fn (?*anyopaque, File) void,
197 fileDowngradeLock: *const fn (?*anyopaque, File) File.DowngradeLockError!void,
198 fileRealPath: *const fn (?*anyopaque, File, out_buffer: []u8) File.RealPathError!usize,
199 fileHardLink: *const fn (?*anyopaque, File, Dir, []const u8, File.HardLinkOptions) File.HardLinkError!void,
200
201 fileMemoryMapCreate: *const fn (?*anyopaque, File, File.MemoryMap.CreateOptions) File.MemoryMap.CreateError!File.MemoryMap,
202 fileMemoryMapDestroy: *const fn (?*anyopaque, *File.MemoryMap) void,
203 fileMemoryMapSetLength: *const fn (?*anyopaque, *File.MemoryMap, usize) File.MemoryMap.SetLengthError!void,
204 fileMemoryMapRead: *const fn (?*anyopaque, *File.MemoryMap) File.ReadPositionalError!void,
205 fileMemoryMapWrite: *const fn (?*anyopaque, *File.MemoryMap) File.WritePositionalError!void,
206
207 processExecutableOpen: *const fn (?*anyopaque, Dir.OpenFileOptions) std.process.OpenExecutableError!File,
208 processExecutablePath: *const fn (?*anyopaque, buffer: []u8) std.process.ExecutablePathError!usize,
209 lockStderr: *const fn (?*anyopaque, ?Terminal.Mode) Cancelable!LockedStderr,
210 tryLockStderr: *const fn (?*anyopaque, ?Terminal.Mode) Cancelable!?LockedStderr,
211 unlockStderr: *const fn (?*anyopaque) void,
212 processCurrentPath: *const fn (?*anyopaque, buffer: []u8) std.process.CurrentPathError!usize,
213 processSetCurrentDir: *const fn (?*anyopaque, Dir) std.process.SetCurrentDirError!void,
214 processSetCurrentPath: *const fn (?*anyopaque, []const u8) std.process.SetCurrentPathError!void,
215 processReplace: *const fn (?*anyopaque, std.process.ReplaceOptions) std.process.ReplaceError,
216 processReplacePath: *const fn (?*anyopaque, Dir, std.process.ReplaceOptions) std.process.ReplaceError,
217 processSpawn: *const fn (?*anyopaque, std.process.SpawnOptions) std.process.SpawnError!std.process.Child,
218 processSpawnPath: *const fn (?*anyopaque, Dir, std.process.SpawnOptions) std.process.SpawnError!std.process.Child,
219 childWait: *const fn (?*anyopaque, *std.process.Child) std.process.Child.WaitError!std.process.Child.Term,
220 childKill: *const fn (?*anyopaque, *std.process.Child) void,
221
222 progressParentFile: *const fn (?*anyopaque) std.Progress.ParentFileError!File,
223
224 now: *const fn (?*anyopaque, Clock) Timestamp,
225 clockResolution: *const fn (?*anyopaque, Clock) Clock.ResolutionError!Duration,
226 sleep: *const fn (?*anyopaque, Timeout) Cancelable!void,
227
228 random: *const fn (?*anyopaque, buffer: []u8) void,
229 randomSecure: *const fn (?*anyopaque, buffer: []u8) RandomSecureError!void,
230
231 netListenIp: *const fn (?*anyopaque, address: *const net.IpAddress, net.IpAddress.ListenOptions) net.IpAddress.ListenError!net.Socket,
232 netAccept: *const fn (?*anyopaque, server: net.Socket.Handle, options: net.Server.AcceptOptions) net.Server.AcceptError!net.Socket,
233 netBindIp: *const fn (?*anyopaque, address: *const net.IpAddress, options: net.IpAddress.BindOptions) net.IpAddress.BindError!net.Socket,
234 netConnectIp: *const fn (?*anyopaque, address: *const net.IpAddress, options: net.IpAddress.ConnectOptions) net.IpAddress.ConnectError!net.Socket,
235 netListenUnix: *const fn (?*anyopaque, *const net.UnixAddress, net.UnixAddress.ListenOptions) net.UnixAddress.ListenError!net.Socket.Handle,
236 netConnectUnix: *const fn (?*anyopaque, *const net.UnixAddress) net.UnixAddress.ConnectError!net.Socket.Handle,
237 netSocketCreatePair: *const fn (?*anyopaque, net.Socket.CreatePairOptions) net.Socket.CreatePairError![2]net.Socket,
238 netWriteFile: *const fn (?*anyopaque, net.Socket.Handle, header: []const u8, *Io.File.Reader, Io.Limit) net.Stream.Writer.WriteFileError!usize,
239 netClose: *const fn (?*anyopaque, sockets: []const net.Socket) void,
240 netShutdown: *const fn (?*anyopaque, handle: net.Socket.Handle, how: net.ShutdownHow) net.ShutdownError!void,
241 netInterfaceNameResolve: *const fn (?*anyopaque, *const net.Interface.Name) net.Interface.Name.ResolveError!net.Interface,
242 netInterfaceName: *const fn (?*anyopaque, net.Interface) net.Interface.NameError!net.Interface.Name,
243 netLookup: *const fn (?*anyopaque, net.HostName, *Queue(net.HostName.LookupResult), net.HostName.LookupOptions) net.HostName.LookupError!void,
244};
245
246pub const Operation = union(enum) {
247 file_read_streaming: FileReadStreaming,
248 file_write_streaming: FileWriteStreaming,
249 /// On Windows this is NtDeviceIoControlFile. On POSIX this is ioctl. On
250 /// other systems this tag is unreachable.
251 device_io_control: DeviceIoControl,
252 net_receive: NetReceive,
253 net_send: NetSend,
254 net_read: NetRead,
255 net_write: NetWrite,
256
257 pub const Tag = @typeInfo(Operation).@"union".tag_type.?;
258
259 /// May return 0 reads which is different than `error.EndOfStream`.
260 pub const FileReadStreaming = struct {
261 file: File,
262 data: []const []u8,
263
264 pub const Error = UnendingError || error{EndOfStream};
265 pub const UnendingError = error{
266 InputOutput,
267 SystemResources,
268 /// Trying to read a directory file descriptor as if it were a file.
269 IsDir,
270 ConnectionResetByPeer,
271 /// File was not opened with read capability.
272 NotOpenForReading,
273 SocketUnconnected,
274 /// Non-blocking has been enabled, and reading from the file descriptor
275 /// would block.
276 WouldBlock,
277 /// In WASI, this error occurs when the file descriptor does
278 /// not hold the required rights to read from it.
279 AccessDenied,
280 /// Unable to read file due to lock. Depending on the `Io` implementation,
281 /// reading from a locked file may return this error, or may ignore the
282 /// lock.
283 LockViolation,
284 } || Io.UnexpectedError;
285
286 pub const Result = Error!usize;
287 };
288
289 pub const FileWriteStreaming = struct {
290 file: File,
291 header: []const u8 = &.{},
292 data: []const []const u8,
293 splat: usize = 1,
294
295 pub const Error = error{
296 DiskQuota,
297 FileTooBig,
298 InputOutput,
299 NoSpaceLeft,
300 DeviceBusy,
301 /// File descriptor does not hold the required rights to write to it.
302 AccessDenied,
303 PermissionDenied,
304 /// File is an unconnected socket, or closed its read end.
305 BrokenPipe,
306 /// Insufficient kernel memory to read from in_fd.
307 SystemResources,
308 NotOpenForWriting,
309 /// The process cannot access the file because another process has locked
310 /// a portion of the file. Windows-only.
311 LockViolation,
312 /// Non-blocking has been enabled and this operation would block.
313 WouldBlock,
314 /// This error occurs when a device gets disconnected before or mid-flush
315 /// while it's being written to - errno(6): No such device or address.
316 NoDevice,
317 FileBusy,
318 } || Io.UnexpectedError;
319
320 pub const Result = Error!usize;
321 };
322
323 pub const DeviceIoControl = switch (builtin.os.tag) {
324 .wasi => noreturn,
325 .windows => struct {
326 file: File,
327 code: std.os.windows.CTL_CODE,
328 in: []const u8 = &.{},
329 out: []u8 = &.{},
330
331 pub const Result = std.os.windows.IO_STATUS_BLOCK;
332 },
333 else => struct {
334 file: File,
335 /// Device-dependent operation code.
336 code: u32,
337 arg: ?*anyopaque,
338
339 /// Device and operation dependent result. Negative values are
340 /// negative errno.
341 pub const Result = i32;
342 },
343 };
344
345 pub const NetReceive = struct {
346 socket_handle: net.Socket.Handle,
347 message_buffer: []net.IncomingMessage,
348 data_buffer: []u8,
349 flags: net.ReceiveFlags,
350
351 pub const Error = error{
352 /// Insufficient memory or other resource internal to the operating system.
353 SystemResources,
354 /// Per-process limit on the number of open file descriptors has been reached.
355 ProcessFdQuotaExceeded,
356 /// System-wide limit on the total number of open files has been reached.
357 SystemFdQuotaExceeded,
358 /// Local end has been shut down on a connection-oriented socket, or
359 /// the socket was never connected.
360 SocketUnconnected,
361 /// The socket type requires that message be sent atomically, and the
362 /// size of the message to be sent made this impossible. The message
363 /// was not transmitted, or was partially transmitted.
364 MessageOversize,
365 /// Network connection was unexpectedly closed by sender.
366 ConnectionResetByPeer,
367 /// The local network interface used to reach the destination is offline.
368 NetworkDown,
369 /// A connectionless packet was previously sent successfully,
370 /// however, it was not received because no service is operating at
371 /// the destination port of the transport on the remote system.
372 /// This caused an ICMP port unreachable packet to be returned to
373 /// the OS where it was queued up to be reported at the next call
374 /// to send or receive on the bound socket.
375 PortUnreachable,
376 /// The remote peer did not respond to ongoing communication, causing the
377 /// OS to abort the connection.
378 ConnectionTimedOut,
379 } || Io.UnexpectedError;
380
381 pub const Result = struct { ?net.Socket.ReceiveError, usize };
382 };
383
384 pub const NetSend = struct {
385 socket_handle: net.Socket.Handle,
386 messages: []net.OutgoingMessage,
387 flags: net.SendFlags,
388
389 pub const Error = error{
390 /// The socket type requires that message be sent atomically, and the
391 /// size of the message to be sent made this impossible. The message
392 /// was not transmitted, or was partially transmitted.
393 MessageOversize,
394 /// The output queue for a network interface was full. This generally indicates that the
395 /// interface has stopped sending, but may be caused by transient congestion. (Normally,
396 /// this does not occur in Linux. Packets are just silently dropped when a device queue
397 /// overflows.)
398 ///
399 /// This is also caused when there is not enough kernel memory available.
400 SystemResources,
401 /// No route to network.
402 NetworkUnreachable,
403 /// Network reached but no route to host.
404 HostUnreachable,
405 /// The local network interface used to reach the destination is offline.
406 NetworkDown,
407 /// The destination address is not listening. Can still occur for
408 /// connectionless messages.
409 ConnectionRefused,
410 /// Operating system or protocol does not support the address family.
411 AddressFamilyUnsupported,
412 /// Another TCP Fast Open is already in progress.
413 FastOpenAlreadyInProgress,
414 /// Network session was unexpectedly closed by recipient.
415 ConnectionResetByPeer,
416 /// Local end has been shut down on a connection-oriented socket, or
417 /// the socket was never connected.
418 SocketUnconnected,
419 /// An attempt was made to send to a network/broadcast address as
420 /// though it was a unicast address.
421 AccessDenied,
422 /// The remote peer did not respond to ongoing communication, causing the
423 /// OS to abort the connection.
424 ConnectionTimedOut,
425 } || Io.UnexpectedError;
426
427 pub const Result = struct { ?net.Socket.SendError, usize };
428 };
429
430 pub const NetRead = struct {
431 socket_handle: net.Socket.Handle,
432 data: [][]u8,
433
434 pub const Error = error{
435 SystemResources,
436 ConnectionResetByPeer,
437 SocketUnconnected,
438 /// File descriptor does not hold the required rights to read from it.
439 AccessDenied,
440 NetworkDown,
441 /// The remote peer did not respond to ongoing communication, causing the
442 /// OS to abort the connection.
443 ConnectionTimedOut,
444 } || Io.UnexpectedError;
445
446 pub const Result = Error!usize;
447 };
448
449 pub const NetWrite = struct {
450 socket_handle: net.Socket.Handle,
451 header: []const u8 = &.{},
452 data: []const []const u8,
453 splat: usize = 1,
454
455 pub const Error = error{
456 /// Another TCP Fast Open is already in progress.
457 FastOpenAlreadyInProgress,
458 /// Network session was unexpectedly closed by recipient.
459 ConnectionResetByPeer,
460 /// The output queue for a network interface was full. This generally indicates that the
461 /// interface has stopped sending, but may be caused by transient congestion. (Normally,
462 /// this does not occur in Linux. Packets are just silently dropped when a device queue
463 /// overflows.)
464 ///
465 /// This is also caused when there is not enough kernel memory available.
466 SystemResources,
467 /// No route to network.
468 NetworkUnreachable,
469 /// Network reached but no route to host.
470 HostUnreachable,
471 /// The local network interface used to reach the destination is down.
472 NetworkDown,
473 /// The destination address is not listening.
474 ConnectionRefused,
475 /// The passed address didn't have the correct address family in its sa_family field.
476 AddressFamilyUnsupported,
477 /// Local end has been shut down on a connection-oriented socket, or
478 /// the socket was never connected.
479 SocketUnconnected,
480 /// The remote peer did not respond to ongoing communication, causing the
481 /// OS to abort the connection.
482 ConnectionTimedOut,
483 SocketNotBound,
484 } || Io.UnexpectedError;
485
486 pub const Result = Error!usize;
487 };
488
489 pub const Result = Result: {
490 const operation_info = @typeInfo(Operation).@"union";
491 const operation_count = operation_info.field_names.len;
492 var field_names: [operation_count][]const u8 = undefined;
493 var field_types: [operation_count]type = undefined;
494 for (operation_info.field_names, operation_info.field_types, &field_names, &field_types) |f_name, f_type, *field_name, *field_type| {
495 field_name.* = f_name;
496 field_type.* = if (f_type == noreturn) noreturn else f_type.Result;
497 }
498 break :Result @Union(.auto, Tag, &field_names, &field_types, &@splat(.{}));
499 };
500
501 pub const Storage = union {
502 unused: List.DoubleNode,
503 submission: Submission,
504 pending: Pending,
505 completion: Completion,
506
507 pub const Submission = struct {
508 node: List.SingleNode,
509 operation: Operation,
510 };
511
512 pub const Pending = struct {
513 node: List.DoubleNode,
514 tag: Tag,
515 userdata: Userdata align(@max(@alignOf(usize), 4)),
516
517 pub const Userdata = [7]usize;
518 };
519
520 pub const Completion = struct {
521 node: List.SingleNode,
522 result: Result,
523 };
524 };
525
526 pub const OptionalIndex = enum(u32) {
527 none = std.math.maxInt(u32),
528 _,
529
530 pub fn fromIndex(i: usize) OptionalIndex {
531 const oi: OptionalIndex = @fromBackingInt(@intCast(i));
532 assert(oi != .none);
533 return oi;
534 }
535
536 pub fn toIndex(oi: OptionalIndex) u32 {
537 assert(oi != .none);
538 return @backingInt(oi);
539 }
540 };
541 pub const List = struct {
542 head: OptionalIndex,
543 tail: OptionalIndex,
544
545 pub const empty: List = .{ .head = .none, .tail = .none };
546
547 pub const SingleNode = struct { next: OptionalIndex };
548 pub const DoubleNode = struct { prev: OptionalIndex, next: OptionalIndex };
549 };
550};
551
552/// Performs one `Operation`.
553pub fn operate(io: Io, operation: Operation) Cancelable!Operation.Result {
554 return io.vtable.operate(io.userdata, operation);
555}
556
557pub const OperateTimeoutError = Cancelable || Timeout.Error || ConcurrentError;
558
559/// Performs one `Operation` with provided `timeout`.
560pub fn operateTimeout(io: Io, operation: Operation, timeout: Timeout) OperateTimeoutError!Operation.Result {
561 if (timeout == .none) return io.vtable.operate(io.userdata, operation);
562 var storage: [1]Operation.Storage = undefined;
563 var batch: Batch = .init(&storage);
564 batch.addAt(0, operation);
565 try batch.awaitConcurrent(io, timeout);
566 const completion = batch.next().?;
567 assert(completion.index == 0);
568 return completion.result;
569}
570
571/// Submits many operations together without waiting for all of them to
572/// complete.
573///
574/// This is a low-level abstraction based on `Operation`. For a higher
575/// level API that operates on `Future`, see `Select` and `Group`.
576pub const Batch = struct {
577 storage: []Operation.Storage,
578 unused: Operation.List,
579 submitted: Operation.List,
580 pending: Operation.List,
581 completed: Operation.List,
582 userdata: ?*anyopaque align(@max(@alignOf(?*anyopaque), 4)),
583
584 /// After calling this, it is safe to unconditionally defer a call to
585 /// `cancel`. `storage` is a pre-allocated buffer of undefined memory that
586 /// determines the maximum number of active operations that can be
587 /// submitted via `add` and `addAt`.
588 pub fn init(storage: []Operation.Storage) Batch {
589 var prev: Operation.OptionalIndex = .none;
590 for (storage, 0..) |*operation, index| {
591 operation.* = .{ .unused = .{ .prev = prev, .next = .fromIndex(index + 1) } };
592 prev = .fromIndex(index);
593 }
594 storage[storage.len - 1].unused.next = .none;
595 return .{
596 .storage = storage,
597 .unused = .{
598 .head = .fromIndex(0),
599 .tail = .fromIndex(storage.len - 1),
600 },
601 .submitted = .empty,
602 .pending = .empty,
603 .completed = .empty,
604 .userdata = null,
605 };
606 }
607
608 /// Adds an operation to be performed at the next await call.
609 /// Returns the index that will be returned by `next` after the operation completes.
610 /// Asserts that no more than `storage.len` operations are active at a time.
611 pub fn add(batch: *Batch, operation: Operation) u32 {
612 const index = batch.unused.head.toIndex();
613 batch.addAt(index, operation);
614 return index;
615 }
616
617 /// Adds an operation to be performed at the next await call.
618 /// After the operation completes, `next` will return `index`.
619 /// Asserts that the operation at `index` is not active.
620 pub fn addAt(batch: *Batch, index: u32, operation: Operation) void {
621 const storage = &batch.storage[index];
622 const unused = storage.unused;
623 switch (unused.prev) {
624 .none => batch.unused.head = unused.next,
625 else => |prev_index| batch.storage[prev_index.toIndex()].unused.next = unused.next,
626 }
627 switch (unused.next) {
628 .none => batch.unused.tail = unused.prev,
629 else => |next_index| batch.storage[next_index.toIndex()].unused.prev = unused.prev,
630 }
631
632 switch (batch.submitted.tail) {
633 .none => batch.submitted.head = .fromIndex(index),
634 else => |tail_index| batch.storage[tail_index.toIndex()].submission.node.next = .fromIndex(index),
635 }
636 storage.* = .{ .submission = .{ .node = .{ .next = .none }, .operation = operation } };
637 batch.submitted.tail = .fromIndex(index);
638 }
639
640 pub const Completion = struct {
641 /// The element within the provided operation storage that completed.
642 /// `addAt` can be used to re-arm the `Batch` using this `index`.
643 index: u32,
644 /// The return value of the operation.
645 result: Operation.Result,
646 };
647
648 /// After calling `awaitAsync`, `awaitConcurrent`, or `cancel`, this
649 /// function iterates over the completed operations.
650 ///
651 /// Each completion returned from this function dequeues from the `Batch`.
652 /// It is not required to dequeue all completions before awaiting again.
653 pub fn next(batch: *Batch) ?Completion {
654 const index = batch.completed.head;
655 if (index == .none) return null;
656 const storage = &batch.storage[index.toIndex()];
657 const completion = storage.completion;
658 const next_index = completion.node.next;
659 batch.completed.head = next_index;
660 if (next_index == .none) batch.completed.tail = .none;
661
662 const tail_index = batch.unused.tail;
663 switch (tail_index) {
664 .none => batch.unused.head = index,
665 else => batch.storage[tail_index.toIndex()].unused.next = index,
666 }
667 storage.* = .{ .unused = .{ .prev = tail_index, .next = .none } };
668 batch.unused.tail = index;
669 return .{ .index = index.toIndex(), .result = completion.result };
670 }
671
672 /// Waits for at least one of the submitted operations to complete. After
673 /// this function returns the completed operations can be iterated with
674 /// `next`.
675 ///
676 /// This function provides opportunity for the implementation to introduce
677 /// concurrency into the batched operations, but unlike `awaitConcurrent`,
678 /// does not require it, and therefore cannot fail with
679 /// `error.ConcurrencyUnavailable`.
680 pub fn awaitAsync(batch: *Batch, io: Io) Cancelable!void {
681 return io.vtable.batchAwaitAsync(io.userdata, batch);
682 }
683
684 pub const AwaitConcurrentError = ConcurrentError || Cancelable || Timeout.Error;
685
686 /// Waits for at least one of the submitted operations to complete. After
687 /// this function returns the completed operations can be iterated with
688 /// `next`.
689 ///
690 /// Unlike `awaitAsync`, this function requires the implementation to
691 /// perform the operations concurrently and therefore can fail with
692 /// `error.ConcurrencyUnavailable`.
693 pub fn awaitConcurrent(batch: *Batch, io: Io, timeout: Timeout) AwaitConcurrentError!void {
694 return io.vtable.batchAwaitConcurrent(io.userdata, batch, timeout);
695 }
696
697 /// Requests all pending operations to be interrupted, then waits for all
698 /// pending operations to complete. After this returns, the `Batch` is in a
699 /// well-defined state, ready to be iterated with `next`. Successfully
700 /// canceled operations will be absent from the iteration. Some operations
701 /// may have successfully completed regardless of the cancel request and
702 /// will appear in the iteration.
703 pub fn cancel(batch: *Batch, io: Io) void {
704 { // abort pending submissions
705 var tail_index = batch.unused.tail;
706 defer batch.unused.tail = tail_index;
707 var index = batch.submitted.head;
708 errdefer batch.submissions.head = index;
709 while (index != .none) {
710 const next_index = batch.storage[index.toIndex()].submission.node.next;
711 switch (tail_index) {
712 .none => batch.unused.head = index,
713 else => batch.storage[tail_index.toIndex()].unused.next = index,
714 }
715 batch.storage[index.toIndex()] = .{ .unused = .{ .prev = tail_index, .next = .none } };
716 tail_index = index;
717 index = next_index;
718 }
719 batch.submitted = .{ .head = .none, .tail = .none };
720 }
721 io.vtable.batchCancel(io.userdata, batch);
722 assert(batch.submitted.head == .none and batch.submitted.tail == .none);
723 assert(batch.pending.head == .none and batch.pending.tail == .none);
724 assert(batch.userdata == null); // that was the last chance to deallocate resources
725 }
726};
727
728pub const Limit = enum(usize) {
729 nothing = 0,
730 unlimited = math.maxInt(usize),
731 _,
732
733 /// `math.maxInt(usize)` is interpreted to mean `.unlimited`.
734 pub fn limited(n: usize) Limit {
735 return @fromBackingInt(@intCast(n));
736 }
737
738 /// Any value grater than `math.maxInt(usize)` is interpreted to mean
739 /// `.unlimited`.
740 pub fn limited64(n: u64) Limit {
741 return @fromBackingInt(@intCast(@min(n, math.maxInt(usize))));
742 }
743
744 pub fn countVec(data: []const []const u8) Limit {
745 var total: usize = 0;
746 for (data) |d| total += d.len;
747 return .limited(total);
748 }
749
750 pub fn min(a: Limit, b: Limit) Limit {
751 return @fromBackingInt(@intCast(@min(@backingInt(a), @backingInt(b))));
752 }
753
754 pub fn max(a: Limit, b: Limit) Limit {
755 if (a == .unlimited or b == .unlimited) {
756 return .unlimited;
757 }
758
759 return @fromBackingInt(@intCast(@max(@backingInt(a), @backingInt(b))));
760 }
761
762 pub fn minInt(l: Limit, n: usize) usize {
763 return @min(n, @backingInt(l));
764 }
765
766 pub fn minInt64(l: Limit, n: u64) usize {
767 return @min(n, @backingInt(l));
768 }
769
770 pub fn slice(l: Limit, s: []u8) []u8 {
771 return s[0..l.minInt(s.len)];
772 }
773
774 pub fn sliceConst(l: Limit, s: []const u8) []const u8 {
775 return s[0..l.minInt(s.len)];
776 }
777
778 pub fn toInt(l: Limit) ?usize {
779 return switch (l) {
780 else => @backingInt(l),
781 .unlimited => null,
782 };
783 }
784
785 pub fn toInt64(l: Limit) ?u64 {
786 return switch (l) {
787 else => @backingInt(l),
788 .unlimited => null,
789 };
790 }
791
792 /// Reduces a slice to account for the limit, leaving room for one extra
793 /// byte above the limit, allowing for the use case of differentiating
794 /// between end-of-stream and reaching the limit.
795 pub fn slice1(l: Limit, non_empty_buffer: []u8) []u8 {
796 assert(non_empty_buffer.len >= 1);
797 return non_empty_buffer[0..@min(@backingInt(l) +| 1, non_empty_buffer.len)];
798 }
799
800 pub fn nonzero(l: Limit) bool {
801 return l != .nothing;
802 }
803
804 /// Return a new limit reduced by `amount` or return `null` indicating
805 /// limit would be exceeded.
806 pub fn subtract(l: Limit, amount: usize) ?Limit {
807 if (l == .unlimited) return .unlimited;
808 if (amount > @backingInt(l)) return null;
809 return @fromBackingInt(@intCast(@backingInt(l) - amount));
810 }
811};
812
813pub const Cancelable = error{
814 /// Caller has requested the async operation to stop.
815 Canceled,
816};
817
818pub const UnexpectedError = error{
819 /// The Operating System returned an undocumented error code.
820 ///
821 /// This error is in theory not possible, but it would be better
822 /// to handle this error than to invoke undefined behavior.
823 ///
824 /// When this error code is observed, it usually means the Zig Standard
825 /// Library needs a small patch to add the error code to the error set for
826 /// the respective function.
827 Unexpected,
828};
829
830pub const Clock = enum {
831 /// A settable system-wide clock that measures real (i.e. wall-clock)
832 /// time. This clock is affected by discontinuous jumps in the system
833 /// time (e.g., if the system administrator manually changes the
834 /// clock), and by frequency adjustments performed by NTP and similar
835 /// applications.
836 ///
837 /// This clock normally counts the number of seconds since 1970-01-01
838 /// 00:00:00 Coordinated Universal Time (UTC) except that it ignores
839 /// leap seconds; near a leap second it is typically adjusted by NTP to
840 /// stay roughly in sync with UTC.
841 ///
842 /// Timestamps returned by implementations of this clock represent time
843 /// elapsed since 1970-01-01T00:00:00Z, the POSIX/Unix epoch, ignoring
844 /// leap seconds. This is colloquially known as "Unix time". If the
845 /// underlying OS uses a different epoch for native timestamps (e.g.,
846 /// Windows, which uses 1601-01-01) they are translated accordingly.
847 real,
848 /// A nonsettable system-wide clock that represents time since some
849 /// unspecified point in the past.
850 ///
851 /// Monotonic: Guarantees that the time returned by consecutive calls
852 /// will not go backwards, but successive calls may return identical
853 /// (not-increased) time values.
854 ///
855 /// Not affected by discontinuous jumps in the system time (e.g., if
856 /// the system administrator manually changes the clock), but may be
857 /// affected by frequency adjustments.
858 ///
859 /// This clock expresses intent to **exclude time that the system is
860 /// suspended**. However, implementations may be unable to satisify
861 /// this, and may include that time.
862 ///
863 /// * On Linux, corresponds `CLOCK_MONOTONIC`.
864 /// * On macOS, corresponds to `CLOCK_UPTIME_RAW`.
865 awake,
866 /// Identical to `awake` except it expresses intent to **include time
867 /// that the system is suspended**, however, due to limitations it may
868 /// behave identically to `awake`.
869 ///
870 /// * On Linux, corresponds `CLOCK_BOOTTIME`.
871 /// * On macOS, corresponds to `CLOCK_MONOTONIC_RAW`.
872 boot,
873 /// Tracks the amount of CPU time in user or kernel mode used by the calling
874 /// process.
875 cpu_process,
876 /// Tracks the amount of CPU time in user or kernel mode used by the calling
877 /// thread.
878 cpu_thread,
879
880 /// This function is not cancelable because it does not block.
881 ///
882 /// Resolution is determined by `resolution` which may be 0 if the
883 /// clock is unsupported.
884 ///
885 /// See also:
886 /// * `Clock.Timestamp.now`
887 pub fn now(clock: Clock, io: Io) Io.Timestamp {
888 return io.vtable.now(io.userdata, clock);
889 }
890
891 pub const ResolutionError = error{
892 ClockUnavailable,
893 Unexpected,
894 };
895
896 /// Reveals the granularity of `clock`. May be zero, indicating
897 /// unsupported clock.
898 pub fn resolution(clock: Clock, io: Io) ResolutionError!Io.Duration {
899 return io.vtable.clockResolution(io.userdata, clock);
900 }
901
902 pub const Timestamp = struct {
903 raw: Io.Timestamp,
904 clock: Clock,
905
906 /// This function is not cancelable because it does not block.
907 ///
908 /// Resolution is determined by `resolution` which may be 0 if
909 /// the clock is unsupported.
910 ///
911 /// See also:
912 /// * `Clock.now`
913 pub fn now(io: Io, clock: Clock) Clock.Timestamp {
914 return .{
915 .raw = io.vtable.now(io.userdata, clock),
916 .clock = clock,
917 };
918 }
919
920 /// Sleeps until the timestamp arrives.
921 ///
922 /// See also:
923 /// * `Io.sleep`
924 /// * `Clock.Duration.sleep`
925 /// * `Timeout.sleep`
926 pub fn wait(t: Clock.Timestamp, io: Io) Cancelable!void {
927 return io.vtable.sleep(io.userdata, .{ .deadline = t });
928 }
929
930 pub fn durationTo(from: Clock.Timestamp, to: Clock.Timestamp) Clock.Duration {
931 assert(from.clock == to.clock);
932 return .{
933 .raw = from.raw.durationTo(to.raw),
934 .clock = from.clock,
935 };
936 }
937
938 pub fn addDuration(from: Clock.Timestamp, duration: Clock.Duration) Clock.Timestamp {
939 assert(from.clock == duration.clock);
940 return .{
941 .raw = from.raw.addDuration(duration.raw),
942 .clock = from.clock,
943 };
944 }
945
946 pub fn subDuration(from: Clock.Timestamp, duration: Clock.Duration) Clock.Timestamp {
947 assert(from.clock == duration.clock);
948 return .{
949 .raw = from.raw.subDuration(duration.raw),
950 .clock = from.clock,
951 };
952 }
953
954 /// Resolution is determined by `resolution` which may be 0 if
955 /// the clock is unsupported.
956 pub fn fromNow(io: Io, duration: Clock.Duration) Clock.Timestamp {
957 return .{
958 .clock = duration.clock,
959 .raw = duration.clock.now(io).addDuration(duration.raw),
960 };
961 }
962
963 /// Resolution is determined by `resolution` which may be 0 if
964 /// the clock is unsupported.
965 pub fn untilNow(timestamp: Clock.Timestamp, io: Io) Clock.Duration {
966 const now_ts = Clock.Timestamp.now(io, timestamp.clock);
967 return timestamp.durationTo(now_ts);
968 }
969
970 /// Resolution is determined by `resolution` which may be 0 if
971 /// the clock is unsupported.
972 pub fn durationFromNow(timestamp: Clock.Timestamp, io: Io) Clock.Duration {
973 const now_ts = timestamp.clock.now(io);
974 return .{
975 .clock = timestamp.clock,
976 .raw = now_ts.durationTo(timestamp.raw),
977 };
978 }
979
980 /// Resolution is determined by `resolution` which may be 0 if
981 /// the clock is unsupported.
982 pub fn toClock(t: Clock.Timestamp, io: Io, clock: Clock) Clock.Timestamp {
983 if (t.clock == clock) return t;
984 const now_old = t.clock.now(io);
985 const now_new = clock.now(io);
986 const duration = now_old.durationTo(t.raw);
987 return .{
988 .clock = clock,
989 .raw = now_new.addDuration(duration),
990 };
991 }
992
993 pub fn compare(lhs: Clock.Timestamp, op: math.CompareOperator, rhs: Clock.Timestamp) bool {
994 assert(lhs.clock == rhs.clock);
995 return math.compare(lhs.raw.nanoseconds, op, rhs.raw.nanoseconds);
996 }
997 };
998
999 pub const Duration = struct {
1000 raw: Io.Duration,
1001 clock: Clock,
1002
1003 /// Waits until a specified amount of time has passed on `clock`.
1004 ///
1005 /// See also:
1006 /// * `Io.sleep`
1007 /// * `Clock.Timestamp.wait`
1008 /// * `Timeout.sleep`
1009 pub fn sleep(duration: Clock.Duration, io: Io) Cancelable!void {
1010 return io.vtable.sleep(io.userdata, .{ .duration = duration });
1011 }
1012 };
1013};
1014
1015pub const Timestamp = struct {
1016 nanoseconds: i96,
1017
1018 pub fn now(io: Io, clock: Clock) Io.Timestamp {
1019 return io.vtable.now(io.userdata, clock);
1020 }
1021
1022 pub const zero: Timestamp = .{ .nanoseconds = 0 };
1023
1024 pub fn durationTo(from: Timestamp, to: Timestamp) Duration {
1025 return .{ .nanoseconds = to.nanoseconds - from.nanoseconds };
1026 }
1027
1028 pub fn addDuration(from: Timestamp, duration: Duration) Timestamp {
1029 return .{ .nanoseconds = from.nanoseconds + duration.nanoseconds };
1030 }
1031
1032 pub fn subDuration(from: Timestamp, duration: Duration) Timestamp {
1033 return .{ .nanoseconds = from.nanoseconds - duration.nanoseconds };
1034 }
1035
1036 pub fn withClock(t: Timestamp, clock: Clock) Clock.Timestamp {
1037 return .{ .raw = t, .clock = clock };
1038 }
1039
1040 pub fn fromNanoseconds(x: i96) Timestamp {
1041 return .{ .nanoseconds = x };
1042 }
1043
1044 pub fn toMicroseconds(t: Timestamp) i64 {
1045 return @intCast(@divTrunc(t.nanoseconds, std.time.ns_per_us));
1046 }
1047
1048 pub fn toMilliseconds(t: Timestamp) i64 {
1049 return @intCast(@divTrunc(t.nanoseconds, std.time.ns_per_ms));
1050 }
1051
1052 pub fn toSeconds(t: Timestamp) i64 {
1053 return @intCast(@divTrunc(t.nanoseconds, std.time.ns_per_s));
1054 }
1055
1056 pub fn toNanoseconds(t: Timestamp) i96 {
1057 return t.nanoseconds;
1058 }
1059
1060 pub fn formatNumber(t: Timestamp, w: *std.Io.Writer, n: std.fmt.Number) std.Io.Writer.Error!void {
1061 return w.printInt(t.nanoseconds, n.mode.base() orelse 10, n.case, .{
1062 .precision = n.precision,
1063 .width = n.width,
1064 .alignment = n.alignment,
1065 .fill = n.fill,
1066 });
1067 }
1068
1069 /// Resolution is determined by `Clock.resolution` which may be 0 if
1070 /// the clock is unsupported.
1071 pub fn untilNow(t: Timestamp, io: Io, clock: Clock) Duration {
1072 const now_ts = clock.now(io);
1073 return t.durationTo(now_ts);
1074 }
1075
1076 pub fn compare(lhs: Timestamp, op: math.CompareOperator, rhs: Timestamp) bool {
1077 return math.compare(lhs.nanoseconds, op, rhs.nanoseconds);
1078 }
1079};
1080
1081pub const Duration = struct {
1082 nanoseconds: i96,
1083
1084 pub const zero: Duration = .{ .nanoseconds = 0 };
1085 pub const max: Duration = .{ .nanoseconds = math.maxInt(i96) };
1086
1087 pub fn fromNanoseconds(x: i96) Duration {
1088 return .{ .nanoseconds = x };
1089 }
1090
1091 pub fn fromMicroseconds(x: i64) Duration {
1092 return .{ .nanoseconds = @as(i96, x) * std.time.ns_per_us };
1093 }
1094
1095 pub fn fromMilliseconds(x: i64) Duration {
1096 return .{ .nanoseconds = @as(i96, x) * std.time.ns_per_ms };
1097 }
1098
1099 pub fn fromSeconds(x: i64) Duration {
1100 return .{ .nanoseconds = @as(i96, x) * std.time.ns_per_s };
1101 }
1102
1103 pub fn toMicroseconds(d: Duration) i64 {
1104 return @intCast(@divTrunc(d.nanoseconds, std.time.ns_per_us));
1105 }
1106
1107 pub fn toMilliseconds(d: Duration) i64 {
1108 return @intCast(@divTrunc(d.nanoseconds, std.time.ns_per_ms));
1109 }
1110
1111 pub fn toSeconds(d: Duration) i64 {
1112 return @intCast(@divTrunc(d.nanoseconds, std.time.ns_per_s));
1113 }
1114
1115 pub fn toNanoseconds(d: Duration) i96 {
1116 return d.nanoseconds;
1117 }
1118
1119 /// Write number of nanoseconds according to its signed magnitude:
1120 /// `[#y][#w][#d][#h][#m]#[.###][n|u|m]s`
1121 pub fn format(duration: Duration, w: *Writer) Writer.Error!void {
1122 if (duration.nanoseconds < 0) try w.writeByte('-');
1123 return formatUnsigned(w, @abs(duration.nanoseconds));
1124 }
1125
1126 fn formatUnsigned(w: *Writer, ns: u96) Writer.Error!void {
1127 var ns_remaining = ns;
1128 inline for (.{
1129 .{ .ns = 365 * std.time.ns_per_day, .sep = 'y' },
1130 .{ .ns = std.time.ns_per_week, .sep = 'w' },
1131 .{ .ns = std.time.ns_per_day, .sep = 'd' },
1132 .{ .ns = std.time.ns_per_hour, .sep = 'h' },
1133 .{ .ns = std.time.ns_per_min, .sep = 'm' },
1134 }) |unit| {
1135 if (ns_remaining >= unit.ns) {
1136 const units = ns_remaining / unit.ns;
1137 try w.printInt(units, 10, .lower, .{});
1138 try w.writeByte(unit.sep);
1139 ns_remaining -= units * unit.ns;
1140 if (ns_remaining == 0) return;
1141 }
1142 }
1143
1144 inline for (.{
1145 .{ .ns = std.time.ns_per_s, .sep = "s" },
1146 .{ .ns = std.time.ns_per_ms, .sep = "ms" },
1147 .{ .ns = std.time.ns_per_us, .sep = "us" },
1148 }) |unit| {
1149 const kunits = ns_remaining * 1000 / unit.ns;
1150 if (kunits >= 1000) {
1151 try w.printInt(kunits / 1000, 10, .lower, .{});
1152 const frac = kunits % 1000;
1153 if (frac > 0) {
1154 // Write up to 3 decimal places
1155 var decimal_buf = [_]u8{ '.', 0, 0, 0 };
1156 var inner: Writer = .fixed(decimal_buf[1..]);
1157 inner.printInt(frac, 10, .lower, .{ .fill = '0', .width = 3 }) catch unreachable;
1158 var end: usize = 4;
1159 while (end > 1) : (end -= 1) {
1160 if (decimal_buf[end - 1] != '0') break;
1161 }
1162 try w.writeAll(decimal_buf[0..end]);
1163 }
1164 return w.writeAll(unit.sep);
1165 }
1166 }
1167
1168 try w.printInt(ns_remaining, 10, .lower, .{});
1169 try w.writeAll("ns");
1170 }
1171
1172 test format {
1173 try testFormat("0ns", 0);
1174 try testFormat("1ns", 1);
1175 try testFormat("-1ns", -(1));
1176 try testFormat("999ns", std.time.ns_per_us - 1);
1177 try testFormat("-999ns", -(std.time.ns_per_us - 1));
1178 try testFormat("1us", std.time.ns_per_us);
1179 try testFormat("-1us", -(std.time.ns_per_us));
1180 try testFormat("1.45us", 1450);
1181 try testFormat("-1.45us", -(1450));
1182 try testFormat("1.5us", 3 * std.time.ns_per_us / 2);
1183 try testFormat("-1.5us", -(3 * std.time.ns_per_us / 2));
1184 try testFormat("14.5us", 14500);
1185 try testFormat("-14.5us", -(14500));
1186 try testFormat("145us", 145000);
1187 try testFormat("-145us", -(145000));
1188 try testFormat("999.999us", std.time.ns_per_ms - 1);
1189 try testFormat("-999.999us", -(std.time.ns_per_ms - 1));
1190 try testFormat("1ms", std.time.ns_per_ms + 1);
1191 try testFormat("-1ms", -(std.time.ns_per_ms + 1));
1192 try testFormat("1.5ms", 3 * std.time.ns_per_ms / 2);
1193 try testFormat("-1.5ms", -(3 * std.time.ns_per_ms / 2));
1194 try testFormat("1.11ms", 1110000);
1195 try testFormat("-1.11ms", -(1110000));
1196 try testFormat("1.111ms", 1111000);
1197 try testFormat("-1.111ms", -(1111000));
1198 try testFormat("1.111ms", 1111100);
1199 try testFormat("-1.111ms", -(1111100));
1200 try testFormat("999.999ms", std.time.ns_per_s - 1);
1201 try testFormat("-999.999ms", -(std.time.ns_per_s - 1));
1202 try testFormat("1s", std.time.ns_per_s);
1203 try testFormat("-1s", -(std.time.ns_per_s));
1204 try testFormat("59.999s", std.time.ns_per_min - 1);
1205 try testFormat("-59.999s", -(std.time.ns_per_min - 1));
1206 try testFormat("1m", std.time.ns_per_min);
1207 try testFormat("-1m", -(std.time.ns_per_min));
1208 try testFormat("1h", std.time.ns_per_hour);
1209 try testFormat("-1h", -(std.time.ns_per_hour));
1210 try testFormat("1d", std.time.ns_per_day);
1211 try testFormat("-1d", -(std.time.ns_per_day));
1212 try testFormat("1w", std.time.ns_per_week);
1213 try testFormat("-1w", -(std.time.ns_per_week));
1214 try testFormat("1y", 365 * std.time.ns_per_day);
1215 try testFormat("-1y", -(365 * std.time.ns_per_day));
1216 try testFormat("1y52w23h59m59.999s", 730 * std.time.ns_per_day - 1); // 365d = 52w1d
1217 try testFormat("-1y52w23h59m59.999s", -(730 * std.time.ns_per_day - 1)); // 365d = 52w1d
1218 try testFormat("1y1h1.001s", 365 * std.time.ns_per_day + std.time.ns_per_hour + std.time.ns_per_s + std.time.ns_per_ms);
1219 try testFormat("-1y1h1.001s", -(365 * std.time.ns_per_day + std.time.ns_per_hour + std.time.ns_per_s + std.time.ns_per_ms));
1220 try testFormat("1y1h1s", 365 * std.time.ns_per_day + std.time.ns_per_hour + std.time.ns_per_s + 999 * std.time.ns_per_us);
1221 try testFormat("-1y1h1s", -(365 * std.time.ns_per_day + std.time.ns_per_hour + std.time.ns_per_s + 999 * std.time.ns_per_us));
1222 try testFormat("1y1h999.999us", 365 * std.time.ns_per_day + std.time.ns_per_hour + std.time.ns_per_ms - 1);
1223 try testFormat("-1y1h999.999us", -(365 * std.time.ns_per_day + std.time.ns_per_hour + std.time.ns_per_ms - 1));
1224 try testFormat("1y1h1ms", 365 * std.time.ns_per_day + std.time.ns_per_hour + std.time.ns_per_ms);
1225 try testFormat("-1y1h1ms", -(365 * std.time.ns_per_day + std.time.ns_per_hour + std.time.ns_per_ms));
1226 try testFormat("1y1h1ms", 365 * std.time.ns_per_day + std.time.ns_per_hour + std.time.ns_per_ms + 1);
1227 try testFormat("-1y1h1ms", -(365 * std.time.ns_per_day + std.time.ns_per_hour + std.time.ns_per_ms + 1));
1228 try testFormat("1y1m999ns", 365 * std.time.ns_per_day + std.time.ns_per_min + 999);
1229 try testFormat("-1y1m999ns", -(365 * std.time.ns_per_day + std.time.ns_per_min + 999));
1230 try testFormat("292y24w3d23h47m16.854s", std.math.maxInt(i64));
1231 try testFormat("-292y24w3d23h47m16.854s", std.math.minInt(i64) + 1);
1232 try testFormat("-292y24w3d23h47m16.854s", std.math.minInt(i64));
1233 }
1234
1235 fn testFormat(expected: []const u8, input: i96) !void {
1236 // worst case: "-XXXXXXXXXXXXXyXXwXXdXXhXXmXX.XXXs".len = 34
1237 var buf: [34]u8 = undefined;
1238 var w: Writer = .fixed(&buf);
1239 try w.print("{f}", .{Duration{ .nanoseconds = input }});
1240 try std.testing.expectEqualStrings(expected, w.buffered());
1241 }
1242};
1243
1244/// Declares under what conditions an operation should return `error.Timeout`.
1245pub const Timeout = union(enum) {
1246 /// `.none` will wait forever
1247 none,
1248 duration: Clock.Duration,
1249 deadline: Clock.Timestamp,
1250
1251 pub const Error = error{Timeout};
1252
1253 pub fn toTimestamp(t: Timeout, io: Io) ?Clock.Timestamp {
1254 return switch (t) {
1255 .none => null,
1256 .duration => |d| .fromNow(io, d),
1257 .deadline => |d| d,
1258 };
1259 }
1260
1261 pub fn toDeadline(t: Timeout, io: Io) Timeout {
1262 return switch (t) {
1263 .none => .none,
1264 .duration => |d| .{ .deadline = .fromNow(io, d) },
1265 .deadline => |d| .{ .deadline = d },
1266 };
1267 }
1268
1269 pub fn toDurationFromNow(t: Timeout, io: Io) ?Clock.Duration {
1270 return switch (t) {
1271 .none => null,
1272 .duration => |d| d,
1273 .deadline => |d| d.durationFromNow(io),
1274 };
1275 }
1276
1277 /// Waits until the timeout has passed.
1278 ///
1279 /// See also:
1280 /// * `Io.sleep`
1281 /// * `Clock.Duration.sleep`
1282 /// * `Clock.Timestamp.wait`
1283 pub fn sleep(timeout: Timeout, io: Io) Cancelable!void {
1284 return io.vtable.sleep(io.userdata, timeout);
1285 }
1286};
1287
1288pub const AnyFuture = opaque {};
1289
1290pub fn Future(Result: type) type {
1291 return struct {
1292 any_future: ?*AnyFuture,
1293 result: Result,
1294
1295 /// Equivalent to `await` but places a cancelation request. This causes the task to receive
1296 /// `error.Canceled` from its next "cancelation point" (if any). A cancelation point is a
1297 /// call to a function in `Io` which can return `error.Canceled`.
1298 ///
1299 /// After cancelation of a task is requested, only the next cancelation point in that task
1300 /// will return `error.Canceled`: future points will not re-signal the cancelation. As such,
1301 /// it is usually a bug to ignore `error.Canceled`. However, to defer handling cancelation
1302 /// requests, see also `recancel` and `CancelProtection`.
1303 ///
1304 /// Idempotent. Not threadsafe.
1305 pub fn cancel(f: *@This(), io: Io) Result {
1306 const any_future = f.any_future orelse return f.result;
1307 io.vtable.cancel(io.userdata, any_future, @ptrCast(&f.result), .of(Result));
1308 f.any_future = null;
1309 return f.result;
1310 }
1311
1312 /// Idempotent. Not threadsafe.
1313 pub fn await(f: *@This(), io: Io) Result {
1314 const any_future = f.any_future orelse return f.result;
1315 io.vtable.await(io.userdata, any_future, @ptrCast(&f.result), .of(Result));
1316 f.any_future = null;
1317 return f.result;
1318 }
1319 };
1320}
1321
1322/// An unordered set of tasks which can only be awaited or canceled as a whole.
1323/// Tasks are spawned in the group with `Group.async` and `Group.concurrent`.
1324///
1325/// The resources associated with each task are *guaranteed* to be released when
1326/// the individual task returns, as opposed to when the whole group completes or
1327/// is awaited. For this reason, it is not a resource leak to have a long-lived
1328/// group which concurrent tasks are repeatedly added to. However, asynchronous
1329/// tasks are not guaranteed to run until `Group.await` or `Group.cancel` is
1330/// called, so adding async tasks to a group without ever awaiting it may leak
1331/// resources.
1332pub const Group = struct {
1333 /// This value indicates whether or not a group has pending tasks. `null`
1334 /// means there are no pending tasks, and no resources associated with the
1335 /// group, so `await` and `cancel` return immediately without calling the
1336 /// implementation. This means that `token` must be accessed atomically to
1337 /// avoid racing with the check in `await` and `cancel`.
1338 token: std.atomic.Value(?*anyopaque),
1339 /// This value is available for the implementation to use as it wishes.
1340 state: usize,
1341
1342 pub const init: Group = .{ .token = .init(null), .state = 0 };
1343
1344 /// Equivalent to `Io.async`, except the task is spawned in this `Group`
1345 /// instead of becoming associated with a `Future`.
1346 ///
1347 /// The return type of `function` must be coercible to `Cancelable!void`.
1348 /// `function` returning `error.Canceled` does nothing because it is an
1349 /// cancelation propagation boundary.
1350 ///
1351 /// Once this function is called, there are resources associated with the
1352 /// group. To release those resources, `await` or `cancel` must eventually
1353 /// be called.
1354 ///
1355 /// `function` is not guaranteed to have been called until `await` or
1356 /// `cancel` is called.
1357 pub fn async(g: *Group, io: Io, function: anytype, args: std.meta.ArgsTuple(@TypeOf(function))) void {
1358 const Args = @TypeOf(args);
1359 const TypeErased = struct {
1360 fn start(context: *const anyopaque) void {
1361 const args_casted: *const Args = @ptrCast(@alignCast(context));
1362 _ = @as(Cancelable!void, @call(.auto, function, args_casted.*)) catch {};
1363 }
1364 };
1365 io.vtable.groupAsync(io.userdata, g, @ptrCast(&args), .of(Args), TypeErased.start);
1366 }
1367
1368 /// Equivalent to `Io.concurrent`, except the task is spawned in this
1369 /// `Group` instead of becoming associated with a `Future`.
1370 ///
1371 /// The return type of `function` must be coercible to `Cancelable!void`.
1372 /// `function` returning `error.Canceled` does nothing because it is an
1373 /// cancelation propagation boundary.
1374 ///
1375 /// Once this function is called, there are resources associated with the
1376 /// group. To release those resources, `Group.await` or `Group.cancel` must
1377 /// eventually be called.
1378 pub fn concurrent(g: *Group, io: Io, function: anytype, args: std.meta.ArgsTuple(@TypeOf(function))) ConcurrentError!void {
1379 const Args = @TypeOf(args);
1380 const TypeErased = struct {
1381 fn start(context: *const anyopaque) void {
1382 const args_casted: *const Args = @ptrCast(@alignCast(context));
1383 _ = @as(Cancelable!void, @call(.auto, function, args_casted.*)) catch {};
1384 }
1385 };
1386 return io.vtable.groupConcurrent(io.userdata, g, @ptrCast(&args), .of(Args), TypeErased.start);
1387 }
1388
1389 /// Blocks until all tasks of the group finish. During this time,
1390 /// cancelation requests propagate to all members of the group, and
1391 /// will also cause `error.Canceled` to be returned when the group
1392 /// does ultimately finish.
1393 ///
1394 /// After this function returns, all tasks of the `Group` created with
1395 /// `async` or `concurrent` are guaranteed to have run.
1396 ///
1397 /// Idempotent. Not threadsafe.
1398 ///
1399 /// It is safe to call this function concurrently with `Group.async` or
1400 /// `Group.concurrent`, provided that the group does not complete until
1401 /// the call to `Group.async` or `Group.concurrent` returns.
1402 pub fn await(g: *Group, io: Io) Cancelable!void {
1403 const token = g.token.load(.acquire) orelse return;
1404 try io.vtable.groupAwait(io.userdata, g, token);
1405 assert(g.token.raw == null);
1406 }
1407
1408 /// Equivalent to `await` but immediately requests cancelation on all
1409 /// members of the group.
1410 ///
1411 /// After this function returns, all tasks of the `Group` created with
1412 /// `async` or `concurrent` are guaranteed to have run.
1413 ///
1414 /// For a description of cancelation and cancelation points, see `Future.cancel`.
1415 ///
1416 /// Idempotent. Not threadsafe.
1417 ///
1418 /// It is safe to call this function concurrently with `Group.async` or
1419 /// `Group.concurrent`, provided that the group does not complete until
1420 /// the call to `Group.async` or `Group.concurrent` returns.
1421 pub fn cancel(g: *Group, io: Io) void {
1422 const token = g.token.load(.acquire) orelse return;
1423 io.vtable.groupCancel(io.userdata, g, token);
1424 assert(g.token.raw == null);
1425 }
1426};
1427
1428/// Asserts that `error.Canceled` was returned from a prior cancelation point, and "re-arms" the
1429/// cancelation request, so that `error.Canceled` will be returned again from the next cancelation
1430/// point.
1431///
1432/// For a description of cancelation and cancelation points, see `Future.cancel`.
1433pub fn recancel(io: Io) void {
1434 io.vtable.recancel(io.userdata);
1435}
1436
1437/// In rare cases, it is desirable to completely block cancelation notification, so that a region
1438/// of code can run uninterrupted before `error.Canceled` is potentially observed. Therefore, every
1439/// task has a "cancel protection" state which indicates whether or not `Io` functions can introduce
1440/// cancelation points.
1441///
1442/// To modify a task's cancel protection state, see `swapCancelProtection`.
1443///
1444/// For a description of cancelation and cancelation points, see `Future.cancel`.
1445pub const CancelProtection = enum(u1) {
1446 /// Any call to an `Io` function with `error.Canceled` in its error set is a cancelation point.
1447 ///
1448 /// This is the default state, which all tasks are created in.
1449 unblocked = 0,
1450 /// No `Io` function introduces a cancelation point (`error.Canceled` will never be returned).
1451 blocked = 1,
1452};
1453/// Updates the current task's cancel protection state (see `CancelProtection`).
1454///
1455/// The typical usage for this function is to protect a block of code from cancelation:
1456/// ```
1457/// const old_cancel_protect = io.swapCancelProtection(.blocked);
1458/// defer _ = io.swapCancelProtection(old_cancel_protect);
1459/// doSomeWork() catch |err| switch (err) {
1460/// error.Canceled => unreachable,
1461/// };
1462/// ```
1463///
1464/// For a description of cancelation and cancelation points, see `Future.cancel`.
1465pub fn swapCancelProtection(io: Io, new: CancelProtection) CancelProtection {
1466 return io.vtable.swapCancelProtection(io.userdata, new);
1467}
1468
1469/// This function acts as a pure cancelation point (subject to protection; see `CancelProtection`)
1470/// and does nothing else. In other words, it returns `error.Canceled` if there is an outstanding
1471/// non-blocked cancelation request, but otherwise is a no-op.
1472///
1473/// It is rarely necessary to call this function. The primary use case is in long-running CPU-bound
1474/// tasks which may need to respond to cancelation before completing. Short tasks, or those which
1475/// perform other `Io` operations (and hence have other cancelation points), will typically already
1476/// respond quickly to cancelation requests.
1477///
1478/// For a description of cancelation and cancelation points, see `Future.cancel`.
1479pub fn checkCancel(io: Io) Cancelable!void {
1480 return io.vtable.checkCancel(io.userdata);
1481}
1482
1483/// Executes tasks together, providing a mechanism to wait until one or more
1484/// tasks complete. Similar to `Batch` but operates at the higher level task
1485/// abstraction layer rather than lower level `Operation` abstraction layer.
1486///
1487/// The provided tagged union will be used as the return type of the await
1488/// function. When calling async or concurrent, one specifies which union field
1489/// the called function's result will be placed into upon completion.
1490pub fn Select(comptime U: type) type {
1491 return struct {
1492 io: Io,
1493 group: Group,
1494 queue: Queue(U),
1495
1496 const S = @This();
1497
1498 pub const Union = U;
1499
1500 pub const Field = std.meta.FieldEnum(U);
1501
1502 pub fn init(io: Io, buffer: []U) S {
1503 return .{
1504 .io = io,
1505 .queue = .init(buffer),
1506 .group = .init,
1507 };
1508 }
1509
1510 /// Calls `function` with `args` asynchronously. The resource spawned is
1511 /// owned by the select.
1512 ///
1513 /// `function` must have return type matching the `field` field of `Union`.
1514 ///
1515 /// `function` *may* be called immediately, before `async` returns.
1516 ///
1517 /// When this function returns, it is guaranteed that `function` has
1518 /// already been called and completed, or it has successfully been
1519 /// assigned a unit of concurrency.
1520 ///
1521 /// After this is called, `await` or `cancel` must be called before the
1522 /// select is deinitialized.
1523 ///
1524 /// Threadsafe.
1525 ///
1526 /// Related:
1527 /// * `Io.async`
1528 /// * `Group.async`
1529 pub fn async(
1530 s: *S,
1531 comptime field: Field,
1532 function: anytype,
1533 args: std.meta.ArgsTuple(@TypeOf(function)),
1534 ) void {
1535 const Context = struct {
1536 select: *S,
1537 args: @TypeOf(args),
1538 fn start(type_erased_context: *const anyopaque) void {
1539 const context: *const @This() = @ptrCast(@alignCast(type_erased_context));
1540 const result = @call(.auto, function, context.args);
1541 const elem = @unionInit(U, @tagName(field), result);
1542 context.select.queue.putOneUncancelable(context.select.io, elem) catch |err| switch (err) {
1543 error.Closed => {},
1544 };
1545 }
1546 };
1547 const context: Context = .{ .select = s, .args = args };
1548 s.io.vtable.groupAsync(s.io.userdata, &s.group, @ptrCast(&context), .of(Context), Context.start);
1549 }
1550
1551 /// Calls `function` with `args` concurrently. The resource spawned is
1552 /// owned by the select.
1553 ///
1554 /// `function` must have return type matching the `field` field of `Union`.
1555 ///
1556 /// After this function returns successfully, it is guaranteed that
1557 /// `function` has been assigned a unit of concurrency, and `await` or
1558 /// `cancel` must be called before the select is deinitialized.
1559 ///
1560 ///
1561 /// Threadsafe.
1562 ///
1563 /// Related:
1564 /// * `Io.concurrent`
1565 /// * `Group.concurrent`
1566 pub fn concurrent(
1567 s: *S,
1568 comptime field: Field,
1569 function: anytype,
1570 args: std.meta.ArgsTuple(@TypeOf(function)),
1571 ) ConcurrentError!void {
1572 const Context = struct {
1573 select: *S,
1574 args: @TypeOf(args),
1575 fn start(type_erased_context: *const anyopaque) void {
1576 const context: *const @This() = @ptrCast(@alignCast(type_erased_context));
1577 const result = @call(.auto, function, context.args);
1578 const elem = @unionInit(U, @tagName(field), result);
1579 context.select.queue.putOneUncancelable(context.select.io, elem) catch |err| switch (err) {
1580 error.Closed => {},
1581 };
1582 }
1583 };
1584 const context: Context = .{ .select = s, .args = args };
1585 try s.io.vtable.groupConcurrent(s.io.userdata, &s.group, @ptrCast(&context), .of(Context), Context.start);
1586 }
1587
1588 /// Blocks until another task of the select finishes.
1589 ///
1590 /// It is legal to call `async` and `concurrent` after this.
1591 ///
1592 /// Threadsafe.
1593 pub fn await(s: *S) Cancelable!U {
1594 return s.queue.getOne(s.io) catch |err| switch (err) {
1595 error.Canceled => |e| return e,
1596 error.Closed => unreachable,
1597 };
1598 }
1599
1600 /// Blocks until at least `min` number of results have been copied
1601 /// into `buffer`.
1602 ///
1603 /// Asserts that `buffer.len >= min`.
1604 ///
1605 /// It is legal to call `async` and `concurrent` after this.
1606 ///
1607 /// Threadsafe.
1608 pub fn awaitMany(s: *S, buffer: []U, min: usize) Cancelable!usize {
1609 return s.queue.get(s.io, buffer, min) catch |err| switch (err) {
1610 error.Canceled => |e| return e,
1611 error.Closed => unreachable,
1612 };
1613 }
1614
1615 /// Requests cancelation on all remaining tasks owned by the select,
1616 /// then blocks until they all finish. If the select was initialized
1617 /// with insufficient buffer space for all remaining tasks to finish, a
1618 /// deadlock occurs.
1619 ///
1620 /// If any of the select tasks allocate resources, those tasks may have
1621 /// completed, meaning that this function must be called in a loop
1622 /// until `null` is returned in order to deallocate those resources. If
1623 /// there is no possibility of resource leaks, `cancelDiscard` is
1624 /// preferable.
1625 ///
1626 /// It is illegal to call `await` or `awaitMany` after this.
1627 ///
1628 /// It is safe to call this multiple times, even after `null` is
1629 /// returned.
1630 ///
1631 /// Threadsafe.
1632 pub fn cancel(s: *S) ?U {
1633 const io = s.io;
1634 s.group.cancel(io);
1635 s.queue.close(io);
1636 return s.queue.getOneUncancelable(io) catch |err| switch (err) {
1637 error.Closed => return null,
1638 };
1639 }
1640
1641 /// Requests cancelation on all remaining tasks owned by the select,
1642 /// then blocks until they all finish.
1643 ///
1644 /// All return values from outstanding tasks are discarded. This
1645 /// function is therefore inappropriate to call when a task can return
1646 /// an allocated resource. For that use case, see `cancel`.
1647 ///
1648 /// It is illegal to call `await` or `awaitMany` after this.
1649 ///
1650 /// It is safe to call this multiple times.
1651 ///
1652 /// Threadsafe.
1653 pub fn cancelDiscard(s: *S) void {
1654 const io = s.io;
1655 const token = s.group.token.load(.acquire) orelse return;
1656 s.queue.close(io);
1657 io.vtable.groupCancel(io.userdata, &s.group, token);
1658 assert(s.group.token.raw == null);
1659 }
1660 };
1661}
1662
1663/// Atomically checks if the value at `ptr` equals `expected`, and if so, blocks until either:
1664///
1665/// * a matching (same `ptr` argument) `futexWake` call occurs, or
1666/// * a spurious ("random") wakeup occurs.
1667///
1668/// Typically, `futexWake` should be called immediately after updating the value at `ptr.*`, to
1669/// unblock tasks using `futexWait` to wait for the value to change from what it previously was.
1670///
1671/// The caller is responsible for identifying spurious wakeups if necessary, typically by checking
1672/// the value at `ptr.*`.
1673///
1674/// Asserts that `T` is 4 bytes in length and has a well-defined layout with no padding bits.
1675pub fn futexWait(io: Io, comptime T: type, ptr: *align(@alignOf(u32)) const T, expected: T) Cancelable!void {
1676 return futexWaitTimeout(io, T, ptr, expected, .none);
1677}
1678/// Same as `futexWait`, except also unblocks if `timeout` expires. As with `futexWait`, spurious
1679/// wakeups are possible. It remains the caller's responsibility to differentiate between these
1680/// three possible wake-up reasons if necessary.
1681pub fn futexWaitTimeout(io: Io, comptime T: type, ptr: *align(@alignOf(u32)) const T, expected: T, timeout: Timeout) Cancelable!void {
1682 const expected_int: u32 = switch (@typeInfo(T)) {
1683 .@"enum" => @bitCast(@backingInt(expected)),
1684 else => @bitCast(expected),
1685 };
1686 return io.vtable.futexWait(io.userdata, @ptrCast(ptr), expected_int, timeout);
1687}
1688/// Same as `futexWait`, except does not introduce a cancelation point.
1689///
1690/// For a description of cancelation and cancelation points, see `Future.cancel`.
1691pub fn futexWaitUncancelable(io: Io, comptime T: type, ptr: *align(@alignOf(u32)) const T, expected: T) void {
1692 const expected_int: u32 = switch (@typeInfo(T)) {
1693 .@"enum" => @bitCast(@backingInt(expected)),
1694 else => @bitCast(expected),
1695 };
1696 io.vtable.futexWaitUncancelable(io.userdata, @ptrCast(ptr), expected_int);
1697}
1698/// Unblocks pending futex waits on `ptr`, up to a limit of `max_waiters` calls.
1699pub fn futexWake(io: Io, comptime T: type, ptr: *align(@alignOf(u32)) const T, max_waiters: u32) void {
1700 comptime assert(@sizeOf(T) == @sizeOf(u32));
1701 if (max_waiters == 0) return;
1702 return io.vtable.futexWake(io.userdata, @ptrCast(ptr), max_waiters);
1703}
1704
1705/// Mutex is a synchronization primitive which enforces atomic access to a
1706/// shared region of code known as the "critical section".
1707///
1708/// Mutex is an extern struct so that it may be used as a field inside another
1709/// extern struct.
1710pub const Mutex = extern struct {
1711 state: std.atomic.Value(State),
1712
1713 pub const init: Mutex = .{ .state = .init(.unlocked) };
1714
1715 pub const State = enum(u32) {
1716 unlocked,
1717 locked_once,
1718 contended,
1719 };
1720
1721 pub fn tryLock(m: *Mutex) bool {
1722 return m.state.cmpxchgStrong(.unlocked, .locked_once, .acquire, .monotonic) == null;
1723 }
1724
1725 pub fn lock(m: *Mutex, io: Io) Cancelable!void {
1726 const initial_state = m.state.cmpxchgStrong(
1727 .unlocked,
1728 .locked_once,
1729 .acquire,
1730 .monotonic,
1731 ) orelse {
1732 @branchHint(.likely);
1733 return;
1734 };
1735 if (initial_state == .contended) {
1736 try io.futexWait(State, &m.state.raw, .contended);
1737 }
1738 while (m.state.swap(.contended, .acquire) != .unlocked) {
1739 try io.futexWait(State, &m.state.raw, .contended);
1740 }
1741 }
1742
1743 /// Same as `lock`, except does not introduce a cancelation point.
1744 ///
1745 /// For a description of cancelation and cancelation points, see `Future.cancel`.
1746 pub fn lockUncancelable(m: *Mutex, io: Io) void {
1747 const initial_state = m.state.cmpxchgStrong(
1748 .unlocked,
1749 .locked_once,
1750 .acquire,
1751 .monotonic,
1752 ) orelse {
1753 @branchHint(.likely);
1754 return;
1755 };
1756 if (initial_state == .contended) {
1757 io.futexWaitUncancelable(State, &m.state.raw, .contended);
1758 }
1759 while (m.state.swap(.contended, .acquire) != .unlocked) {
1760 io.futexWaitUncancelable(State, &m.state.raw, .contended);
1761 }
1762 }
1763
1764 pub fn unlock(m: *Mutex, io: Io) void {
1765 switch (m.state.swap(.unlocked, .release)) {
1766 .unlocked => unreachable,
1767 .locked_once => {},
1768 .contended => {
1769 @branchHint(.unlikely);
1770 io.futexWake(State, &m.state.raw, 1);
1771 },
1772 }
1773 }
1774};
1775
1776pub const Condition = struct {
1777 state: std.atomic.Value(State),
1778 /// Incremented whenever the condition is signaled
1779 epoch: std.atomic.Value(u32),
1780
1781 const State = packed struct(u32) {
1782 waiters: u16,
1783 signals: u16,
1784 };
1785
1786 pub const init: Condition = .{
1787 .state = .init(.{ .waiters = 0, .signals = 0 }),
1788 .epoch = .init(0),
1789 };
1790
1791 /// Blocks until the condition is signaled or canceled.
1792 ///
1793 /// See also:
1794 /// * `waitUncancelable`
1795 /// * `waitTimeout`
1796 pub fn wait(cond: *Condition, io: Io, mutex: *Mutex) Cancelable!void {
1797 waitTimeout(cond, io, mutex, .none) catch |err| switch (err) {
1798 error.Timeout => unreachable,
1799 error.Canceled => |e| return e,
1800 };
1801 }
1802
1803 pub const WaitTimeoutError = Cancelable || Timeout.Error;
1804
1805 /// Blocks until the condition is signaled, canceled, or the provided
1806 /// timeout expires.
1807 ///
1808 /// See also:
1809 /// * `wait`
1810 /// * `waitUncancelable`
1811 pub fn waitTimeout(cond: *Condition, io: Io, mutex: *Mutex, timeout: Timeout) WaitTimeoutError!void {
1812 const deadline = timeout.toDeadline(io);
1813
1814 var epoch = cond.epoch.load(.acquire); // `.acquire` to ensure ordered before state load
1815
1816 {
1817 const prev_state = cond.state.fetchAdd(.{ .waiters = 1, .signals = 0 }, .monotonic);
1818 assert(prev_state.waiters < math.maxInt(u16)); // overflow caused by too many waiters
1819 }
1820
1821 mutex.unlock(io);
1822 defer mutex.lockUncancelable(io);
1823
1824 while (true) {
1825 const result = io.futexWaitTimeout(u32, &cond.epoch.raw, epoch, deadline);
1826
1827 epoch = cond.epoch.load(.acquire); // `.acquire` to ensure ordered before `state` laod
1828
1829 // We were woken normally, so try to consume a pending signal. A signal takes
1830 // priority over an expired deadline, so this is checked before the deadline
1831 // below. On error we safely remove ourselves as a waiter and propagate the error.
1832 if (result) |_| {
1833 var prev_state = cond.state.load(.monotonic);
1834 while (prev_state.signals > 0) {
1835 prev_state = cond.state.cmpxchgWeak(prev_state, .{
1836 .waiters = prev_state.waiters - 1,
1837 .signals = prev_state.signals - 1,
1838 }, .acquire, .monotonic) orelse {
1839 // We successfully consumed a signal.
1840 return;
1841 };
1842 }
1843 } else |err| {
1844 cond.deregister(io);
1845 return err;
1846 }
1847
1848 // There are no signals available and no error; if a timeout was specified and
1849 // the deadline has passed, remove ourselves as a waiter and return
1850 // `error.Timeout`. Otherwise, this was a spurious wakeup: loop back to the
1851 // futex wait.
1852 switch (deadline) {
1853 .none => {},
1854 .deadline => |d| if (d.untilNow(io).raw.nanoseconds >= 0) {
1855 cond.deregister(io);
1856 return error.Timeout;
1857 },
1858 .duration => unreachable,
1859 }
1860 }
1861 }
1862
1863 /// Same as `wait`, except does not introduce a cancelation point.
1864 ///
1865 /// See `Future.cancel` for a description of cancelation points.
1866 pub fn waitUncancelable(cond: *Condition, io: Io, mutex: *Mutex) void {
1867 var epoch = cond.epoch.load(.acquire); // `.acquire` to ensure ordered before state load
1868
1869 {
1870 const prev_state = cond.state.fetchAdd(.{ .waiters = 1, .signals = 0 }, .monotonic);
1871 assert(prev_state.waiters < math.maxInt(u16)); // overflow caused by too many waiters
1872 }
1873
1874 mutex.unlock(io);
1875 defer mutex.lockUncancelable(io);
1876
1877 while (true) {
1878 io.futexWaitUncancelable(u32, &cond.epoch.raw, epoch);
1879
1880 epoch = cond.epoch.load(.acquire); // `.acquire` to ensure ordered before `state` laod
1881
1882 // Even on error, try to consume a pending signal first. Otherwise a race might
1883 // cause a signal to get stuck in the state with no corresponding waiter.
1884 {
1885 var prev_state = cond.state.load(.monotonic);
1886 while (prev_state.signals > 0) {
1887 prev_state = cond.state.cmpxchgWeak(prev_state, .{
1888 .waiters = prev_state.waiters - 1,
1889 .signals = prev_state.signals - 1,
1890 }, .acquire, .monotonic) orelse {
1891 // We successfully consumed a signal.
1892 return;
1893 };
1894 }
1895 }
1896
1897 // There are no more signals available; this was a spurious wakeup,
1898 // so we'll loop back to the futex wait.
1899 }
1900 }
1901
1902 fn deregister(cond: *Condition, io: Io) void {
1903 var prev_state = cond.state.load(.monotonic);
1904 while (true) {
1905 const new_signals = @min(prev_state.signals, prev_state.waiters - 1);
1906 prev_state = cond.state.cmpxchgWeak(prev_state, .{
1907 .waiters = prev_state.waiters - 1,
1908 .signals = new_signals,
1909 }, .monotonic, .monotonic) orelse {
1910 if (prev_state.signals > 0 and prev_state.signals < prev_state.waiters) {
1911 // We kept a signal we are not consuming; wake a remaining waiter for it.
1912 _ = cond.epoch.fetchAdd(1, .release);
1913 io.futexWake(u32, &cond.epoch.raw, 1);
1914 }
1915 return;
1916 };
1917 }
1918 }
1919
1920 pub fn signal(cond: *Condition, io: Io) void {
1921 var prev_state = cond.state.load(.monotonic);
1922 while (prev_state.waiters > prev_state.signals) {
1923 @branchHint(.unlikely);
1924 prev_state = cond.state.cmpxchgWeak(prev_state, .{
1925 .waiters = prev_state.waiters,
1926 .signals = prev_state.signals + 1,
1927 }, .release, .monotonic) orelse {
1928 // Update the epoch to tell the waiting threads that there are new signals for them.
1929 // Note that a waiting thread could miss a take if *exactly* (1<<32)-1 wakes happen
1930 // between it observing the epoch and sleeping on it, but this is extraordinarily
1931 // unlikely due to the precise number of calls required.
1932 _ = cond.epoch.fetchAdd(1, .release); // `.release` to ensure ordered after `state` update
1933 io.futexWake(u32, &cond.epoch.raw, 1);
1934 return;
1935 };
1936 }
1937 }
1938
1939 pub fn broadcast(cond: *Condition, io: Io) void {
1940 var prev_state = cond.state.load(.monotonic);
1941 while (prev_state.waiters > prev_state.signals) {
1942 @branchHint(.unlikely);
1943 prev_state = cond.state.cmpxchgWeak(prev_state, .{
1944 .waiters = prev_state.waiters,
1945 .signals = prev_state.waiters,
1946 }, .release, .monotonic) orelse {
1947 // Update the epoch to tell the waiting threads that there are new signals for them.
1948 // Note that a waiting thread could miss a take if *exactly* (1<<32)-1 wakes happen
1949 // between it observing the epoch and sleeping on it, but this is extraordinarily
1950 // unlikely due to the precise number of calls required.
1951 _ = cond.epoch.fetchAdd(1, .release); // `.release` to ensure ordered after `state` update
1952 io.futexWake(u32, &cond.epoch.raw, prev_state.waiters - prev_state.signals);
1953 return;
1954 };
1955 }
1956 }
1957};
1958
1959/// Logical boolean flag which can be set and unset and supports a "wait until set" operation.
1960pub const Event = enum(u32) {
1961 unset,
1962 waiting,
1963 is_set,
1964
1965 /// Returns whether the logical boolean is `true`.
1966 pub fn isSet(event: *const Event) bool {
1967 return switch (@atomicLoad(Event, event, .acquire)) {
1968 .unset, .waiting => false,
1969 .is_set => true,
1970 };
1971 }
1972
1973 /// Blocks until the logical boolean is `true`.
1974 pub fn wait(event: *Event, io: Io) Cancelable!void {
1975 if (@cmpxchgStrong(Event, event, .unset, .waiting, .acquire, .acquire)) |prev| switch (prev) {
1976 .unset => unreachable,
1977 .waiting => {},
1978 .is_set => return,
1979 };
1980 errdefer {
1981 // Ideally we would restore the event back to `.unset` instead of `.waiting`, but there
1982 // might be other threads waiting on the event. In theory we could track the *number* of
1983 // waiting threads in the unused bits of the `Event`, but that has its own problem: the
1984 // waiters would wake up when a *new waiter* was added. So it's easiest to just leave
1985 // the state at `.waiting`---at worst it causes one redundant call to `futexWake`.
1986 }
1987 while (true) {
1988 try io.futexWait(Event, event, .waiting);
1989 switch (@atomicLoad(Event, event, .acquire)) {
1990 .unset => unreachable, // `reset` called before pending `wait` returned
1991 .waiting => continue,
1992 .is_set => return,
1993 }
1994 }
1995 }
1996
1997 /// Same as `wait`, except does not introduce a cancelation point.
1998 ///
1999 /// For a description of cancelation and cancelation points, see `Future.cancel`.
2000 pub fn waitUncancelable(event: *Event, io: Io) void {
2001 if (@cmpxchgStrong(Event, event, .unset, .waiting, .acquire, .acquire)) |prev| switch (prev) {
2002 .unset => unreachable,
2003 .waiting => {},
2004 .is_set => return,
2005 };
2006 while (true) {
2007 io.futexWaitUncancelable(Event, event, .waiting);
2008 switch (@atomicLoad(Event, event, .acquire)) {
2009 .unset => unreachable, // `reset` called before pending `wait` returned
2010 .waiting => continue,
2011 .is_set => return,
2012 }
2013 }
2014 }
2015
2016 pub const WaitTimeoutError = error{Timeout} || Cancelable;
2017
2018 /// Blocks the calling thread until either the logical boolean is set, the timeout expires, or a
2019 /// spurious wakeup occurs. If the timeout expires or a spurious wakeup occurs, `error.Timeout`
2020 /// is returned.
2021 pub fn waitTimeout(event: *Event, io: Io, timeout: Timeout) WaitTimeoutError!void {
2022 if (@cmpxchgStrong(Event, event, .unset, .waiting, .acquire, .acquire)) |prev| switch (prev) {
2023 .unset => unreachable,
2024 .waiting => {},
2025 .is_set => return,
2026 };
2027 errdefer {
2028 // Ideally we would restore the event back to `.unset` instead of `.waiting`, but there
2029 // might be other threads waiting on the event. In theory we could track the *number* of
2030 // waiting threads in the unused bits of the `Event`, but that has its own problem: the
2031 // waiters would wake up when a *new waiter* was added. So it's easiest to just leave
2032 // the state at `.waiting`---at worst it causes one redundant call to `futexWake`.
2033 }
2034 try io.futexWaitTimeout(Event, event, .waiting, timeout);
2035 switch (@atomicLoad(Event, event, .acquire)) {
2036 .unset => unreachable, // `reset` called before pending `wait` returned
2037 .waiting => return error.Timeout,
2038 .is_set => return,
2039 }
2040 }
2041
2042 /// Sets the logical boolean to true, and hence unblocks any pending calls to `wait`. The
2043 /// logical boolean remains true until `reset` is called, so future calls to `set` have no
2044 /// semantic effect.
2045 ///
2046 /// Any memory accesses prior to a `set` call are "released", so that if this `set` call causes
2047 /// `isSet` to return `true` or a wait to finish, those tasks will be able to observe those
2048 /// memory accesses.
2049 pub fn set(e: *Event, io: Io) void {
2050 switch (@atomicRmw(Event, e, .Xchg, .is_set, .release)) {
2051 .unset, .is_set => {},
2052 .waiting => io.futexWake(Event, e, math.maxInt(u32)),
2053 }
2054 }
2055
2056 /// Sets the logical boolean to false.
2057 ///
2058 /// Assumes that there is no pending call to `wait` or `waitUncancelable`.
2059 ///
2060 /// However, concurrent calls to `isSet`, `set`, and `reset` are allowed.
2061 pub fn reset(e: *Event) void {
2062 @atomicStore(Event, e, .unset, .monotonic);
2063 }
2064};
2065
2066pub const QueueClosedError = error{Closed};
2067
2068pub const TypeErasedQueue = struct {
2069 mutex: Mutex,
2070 closed: bool,
2071
2072 /// Ring buffer. This data is logically *after* queued getters.
2073 buffer: []u8,
2074 start: usize,
2075 len: usize,
2076
2077 putters: std.DoublyLinkedList,
2078 getters: std.DoublyLinkedList,
2079
2080 const Put = struct {
2081 remaining: []const u8,
2082 needed: usize,
2083 condition: Condition,
2084 node: std.DoublyLinkedList.Node,
2085 };
2086
2087 const Get = struct {
2088 remaining: []u8,
2089 needed: usize,
2090 condition: Condition,
2091 node: std.DoublyLinkedList.Node,
2092 };
2093
2094 pub fn init(buffer: []u8) TypeErasedQueue {
2095 return .{
2096 .mutex = .init,
2097 .closed = false,
2098 .buffer = buffer,
2099 .start = 0,
2100 .len = 0,
2101 .putters = .{},
2102 .getters = .{},
2103 };
2104 }
2105
2106 /// After this is called, the queue enters a "closed" state. A closed
2107 /// queue always returns `error.Closed` for put attempts even when
2108 /// there is space in the buffer. However, existing elements of the
2109 /// queue are retrieved before `error.Closed` is returned.
2110 ///
2111 /// Idempotent. Threadsafe.
2112 pub fn close(q: *TypeErasedQueue, io: Io) void {
2113 q.mutex.lockUncancelable(io);
2114 defer q.mutex.unlock(io);
2115 q.closed = true;
2116 {
2117 var it = q.getters.first;
2118 while (it) |node| : (it = node.next) {
2119 const getter: *Get = @alignCast(@fieldParentPtr("node", node));
2120 getter.condition.signal(io);
2121 }
2122 }
2123 {
2124 var it = q.putters.first;
2125 while (it) |node| : (it = node.next) {
2126 const putter: *Put = @alignCast(@fieldParentPtr("node", node));
2127 putter.condition.signal(io);
2128 }
2129 }
2130 }
2131
2132 pub fn put(q: *TypeErasedQueue, io: Io, elements: []const u8, min: usize) (QueueClosedError || Cancelable)!usize {
2133 assert(elements.len >= min);
2134 if (elements.len == 0) return 0;
2135 try q.mutex.lock(io);
2136 defer q.mutex.unlock(io);
2137 return q.putLocked(io, elements, min, false);
2138 }
2139
2140 /// Same as `put`, except does not introduce a cancelation point.
2141 ///
2142 /// For a description of cancelation and cancelation points, see `Future.cancel`.
2143 pub fn putUncancelable(q: *TypeErasedQueue, io: Io, elements: []const u8, min: usize) QueueClosedError!usize {
2144 assert(elements.len >= min);
2145 if (elements.len == 0) return 0;
2146 q.mutex.lockUncancelable(io);
2147 defer q.mutex.unlock(io);
2148 return q.putLocked(io, elements, min, true) catch |err| switch (err) {
2149 error.Canceled => unreachable,
2150 error.Closed => |e| return e,
2151 };
2152 }
2153
2154 fn puttableSlice(q: *const TypeErasedQueue) ?[]u8 {
2155 const unwrapped_index = q.start + q.len;
2156 const wrapped_index, const overflow = @subWithOverflow(unwrapped_index, q.buffer.len);
2157 const slice = switch (overflow) {
2158 1 => q.buffer[unwrapped_index..],
2159 0 => q.buffer[wrapped_index..q.start],
2160 };
2161 return if (slice.len > 0) slice else null;
2162 }
2163
2164 fn putLocked(q: *TypeErasedQueue, io: Io, elements: []const u8, min: usize, uncancelable: bool) (QueueClosedError || Cancelable)!usize {
2165 // A closed queue cannot be added to, even if there is space in the buffer.
2166 if (q.closed) return error.Closed;
2167
2168 // Getters have first priority on the data, and only when the getters
2169 // queue is empty do we start populating the buffer.
2170
2171 // The number of elements we add immediately, before possibly blocking.
2172 var n: usize = 0;
2173
2174 while (q.getters.popFirst()) |getter_node| {
2175 const getter: *Get = @alignCast(@fieldParentPtr("node", getter_node));
2176 const copy_len = @min(getter.remaining.len, elements.len - n);
2177 assert(copy_len > 0);
2178 @memcpy(getter.remaining[0..copy_len], elements[n..][0..copy_len]);
2179 getter.remaining = getter.remaining[copy_len..];
2180 getter.needed -|= copy_len;
2181 n += copy_len;
2182 if (getter.needed == 0) {
2183 getter.condition.signal(io);
2184 } else {
2185 assert(n == elements.len); // we didn't have enough elements for the getter
2186 q.getters.prepend(getter_node);
2187 }
2188 if (n == elements.len) return elements.len;
2189 }
2190
2191 while (q.puttableSlice()) |slice| {
2192 const copy_len = @min(slice.len, elements.len - n);
2193 assert(copy_len > 0);
2194 @memcpy(slice[0..copy_len], elements[n..][0..copy_len]);
2195 q.len += copy_len;
2196 n += copy_len;
2197 if (n == elements.len) return elements.len;
2198 }
2199
2200 // Don't block if we hit the min.
2201 if (n >= min) return n;
2202
2203 var pending: Put = .{
2204 .remaining = elements[n..],
2205 .needed = min - n,
2206 .condition = .init,
2207 .node = .{},
2208 };
2209 q.putters.append(&pending.node);
2210 defer if (pending.needed > 0) q.putters.remove(&pending.node);
2211
2212 while (pending.needed > 0 and !q.closed) {
2213 if (uncancelable) {
2214 pending.condition.waitUncancelable(io, &q.mutex);
2215 continue;
2216 }
2217 pending.condition.wait(io, &q.mutex) catch |err| switch (err) {
2218 error.Canceled => if (pending.remaining.len == elements.len) {
2219 // Canceled while waiting, and appended no elements.
2220 return error.Canceled;
2221 } else {
2222 // Canceled while waiting, but appended some elements, so report those first.
2223 io.recancel();
2224 return elements.len - pending.remaining.len;
2225 },
2226 };
2227 }
2228 if (pending.remaining.len == elements.len) {
2229 // The queue was closed while we were waiting. We appended no elements.
2230 assert(q.closed);
2231 return error.Closed;
2232 }
2233 return elements.len - pending.remaining.len;
2234 }
2235
2236 pub fn get(q: *TypeErasedQueue, io: Io, buffer: []u8, min: usize) (QueueClosedError || Cancelable)!usize {
2237 assert(buffer.len >= min);
2238 if (buffer.len == 0) return 0;
2239 try q.mutex.lock(io);
2240 defer q.mutex.unlock(io);
2241 return q.getLocked(io, buffer, min, false);
2242 }
2243
2244 /// Same as `get`, except does not introduce a cancelation point.
2245 ///
2246 /// For a description of cancelation and cancelation points, see `Future.cancel`.
2247 pub fn getUncancelable(q: *TypeErasedQueue, io: Io, buffer: []u8, min: usize) QueueClosedError!usize {
2248 assert(buffer.len >= min);
2249 if (buffer.len == 0) return 0;
2250 q.mutex.lockUncancelable(io);
2251 defer q.mutex.unlock(io);
2252 return q.getLocked(io, buffer, min, true) catch |err| switch (err) {
2253 error.Canceled => unreachable,
2254 error.Closed => |e| return e,
2255 };
2256 }
2257
2258 fn gettableSlice(q: *const TypeErasedQueue) ?[]const u8 {
2259 const overlong_slice = q.buffer[q.start..];
2260 const slice = overlong_slice[0..@min(overlong_slice.len, q.len)];
2261 return if (slice.len > 0) slice else null;
2262 }
2263
2264 fn getLocked(q: *TypeErasedQueue, io: Io, buffer: []u8, min: usize, uncancelable: bool) (QueueClosedError || Cancelable)!usize {
2265 // The ring buffer gets first priority, then data should come from any
2266 // queued putters, then finally the ring buffer should be filled with
2267 // data from putters so they can be resumed.
2268
2269 // The number of elements we received immediately, before possibly blocking.
2270 var n: usize = 0;
2271
2272 while (q.gettableSlice()) |slice| {
2273 const copy_len = @min(slice.len, buffer.len - n);
2274 assert(copy_len > 0);
2275 @memcpy(buffer[n..][0..copy_len], slice[0..copy_len]);
2276 q.start += copy_len;
2277 if (q.buffer.len - q.start == 0) q.start = 0;
2278 q.len -= copy_len;
2279 n += copy_len;
2280 if (n == buffer.len) {
2281 q.fillRingBufferFromPutters(io);
2282 return buffer.len;
2283 }
2284 }
2285
2286 // Copy directly from putters into buffer.
2287 while (q.putters.popFirst()) |putter_node| {
2288 const putter: *Put = @alignCast(@fieldParentPtr("node", putter_node));
2289 const copy_len = @min(putter.remaining.len, buffer.len - n);
2290 assert(copy_len > 0);
2291 @memcpy(buffer[n..][0..copy_len], putter.remaining[0..copy_len]);
2292 putter.remaining = putter.remaining[copy_len..];
2293 putter.needed -|= copy_len;
2294 n += copy_len;
2295 if (putter.needed == 0) {
2296 putter.condition.signal(io);
2297 } else {
2298 assert(n == buffer.len); // we didn't have enough space for the putter
2299 q.putters.prepend(putter_node);
2300 }
2301 if (n == buffer.len) {
2302 q.fillRingBufferFromPutters(io);
2303 return buffer.len;
2304 }
2305 }
2306
2307 // No need to call `fillRingBufferFromPutters` from this point onwards,
2308 // because we emptied the ring buffer *and* the putter queue!
2309
2310 // Don't block if we hit the min or if the queue is closed. Return how
2311 // many elements we could get immediately, unless the queue was closed and
2312 // empty, in which case report `error.Closed`.
2313 if (n == 0 and q.closed) return error.Closed;
2314 if (n >= min or q.closed) return n;
2315
2316 var pending: Get = .{
2317 .remaining = buffer[n..],
2318 .needed = min - n,
2319 .condition = .init,
2320 .node = .{},
2321 };
2322 q.getters.append(&pending.node);
2323 defer if (pending.needed > 0) q.getters.remove(&pending.node);
2324
2325 while (pending.needed > 0 and !q.closed) {
2326 if (uncancelable) {
2327 pending.condition.waitUncancelable(io, &q.mutex);
2328 continue;
2329 }
2330 pending.condition.wait(io, &q.mutex) catch |err| switch (err) {
2331 error.Canceled => if (pending.remaining.len == buffer.len) {
2332 // Canceled while waiting, and received no elements.
2333 return error.Canceled;
2334 } else {
2335 // Canceled while waiting, but received some elements, so report those first.
2336 io.recancel();
2337 return buffer.len - pending.remaining.len;
2338 },
2339 };
2340 }
2341 if (pending.remaining.len == buffer.len) {
2342 // The queue was closed while we were waiting. We received no elements.
2343 assert(q.closed);
2344 return error.Closed;
2345 }
2346 return buffer.len - pending.remaining.len;
2347 }
2348
2349 /// Called when there is nonzero space available in the ring buffer and
2350 /// potentially putters waiting. The mutex is already held and the task is
2351 /// to copy putter data to the ring buffer and signal any putters whose
2352 /// buffers been fully copied.
2353 fn fillRingBufferFromPutters(q: *TypeErasedQueue, io: Io) void {
2354 while (q.putters.popFirst()) |putter_node| {
2355 const putter: *Put = @alignCast(@fieldParentPtr("node", putter_node));
2356 while (q.puttableSlice()) |slice| {
2357 const copy_len = @min(slice.len, putter.remaining.len);
2358 assert(copy_len > 0);
2359 @memcpy(slice[0..copy_len], putter.remaining[0..copy_len]);
2360 q.len += copy_len;
2361 putter.remaining = putter.remaining[copy_len..];
2362 putter.needed -|= copy_len;
2363 if (putter.needed == 0) {
2364 putter.condition.signal(io);
2365 break;
2366 }
2367 } else {
2368 q.putters.prepend(putter_node);
2369 break;
2370 }
2371 }
2372 }
2373};
2374
2375/// Many producer, many consumer, thread-safe, runtime configurable buffer size.
2376/// When buffer is empty, consumers suspend and are resumed by producers.
2377/// When buffer is full, producers suspend and are resumed by consumers.
2378pub fn Queue(Elem: type) type {
2379 return struct {
2380 type_erased: TypeErasedQueue,
2381
2382 pub fn init(buffer: []Elem) @This() {
2383 return .{ .type_erased = .init(@ptrCast(buffer)) };
2384 }
2385
2386 /// After this is called, the queue enters a "closed" state. A closed
2387 /// queue always returns `error.Closed` for put attempts even when
2388 /// there is space in the buffer. However, existing elements of the
2389 /// queue are retrieved before `error.Closed` is returned.
2390 ///
2391 /// Threadsafe.
2392 pub fn close(q: *@This(), io: Io) void {
2393 q.type_erased.close(io);
2394 }
2395
2396 /// Appends elements to the end of the queue, potentially blocking if
2397 /// there is insufficient capacity. Returns when any one of the
2398 /// following conditions is satisfied:
2399 ///
2400 /// * At least `min` elements have been added to the queue
2401 /// * The queue is closed
2402 /// * The current task is canceled
2403 ///
2404 /// Returns how many of `elements` have been added to the queue, if any.
2405 /// If an error is returned, no elements have been added.
2406 ///
2407 /// If the queue is closed or the task is canceled, but some items were
2408 /// already added before the closure or cancelation, then `put` may
2409 /// return a number lower than `min`, in which case future calls are
2410 /// guaranteed to return `error.Canceled` or `error.Closed`.
2411 ///
2412 /// A return value of 0 is only possible if `min` is 0, in which case
2413 /// the call is guaranteed to queue as many of `elements` as is possible
2414 /// *without* blocking.
2415 ///
2416 /// Asserts that `elements.len >= min`.
2417 pub fn put(q: *@This(), io: Io, elements: []const Elem, min: usize) (QueueClosedError || Cancelable)!usize {
2418 return @divExact(try q.type_erased.put(io, @ptrCast(elements), min * @sizeOf(Elem)), @sizeOf(Elem));
2419 }
2420
2421 /// Same as `put` but blocks until all elements have been added to the queue.
2422 ///
2423 /// If the queue is closed or canceled, `error.Closed` or `error.Canceled`
2424 /// is returned, and it is unspecified how many, if any, of `elements` were
2425 /// added to the queue prior to cancelation or closure.
2426 pub fn putAll(q: *@This(), io: Io, elements: []const Elem) (QueueClosedError || Cancelable)!void {
2427 const n = try q.put(io, elements, elements.len);
2428 if (n != elements.len) {
2429 _ = try q.put(io, elements[n..], elements.len - n);
2430 unreachable; // partial `put` implies queue was closed or we were canceled
2431 }
2432 }
2433
2434 /// Same as `put`, except does not introduce a cancelation point.
2435 ///
2436 /// For a description of cancelation and cancelation points, see `Future.cancel`.
2437 pub fn putUncancelable(q: *@This(), io: Io, elements: []const Elem, min: usize) QueueClosedError!usize {
2438 return @divExact(try q.type_erased.putUncancelable(io, @ptrCast(elements), min * @sizeOf(Elem)), @sizeOf(Elem));
2439 }
2440
2441 /// Appends `item` to the end of the queue, blocking if the queue is full.
2442 pub fn putOne(q: *@This(), io: Io, item: Elem) (QueueClosedError || Cancelable)!void {
2443 assert(try q.put(io, &.{item}, 1) == 1);
2444 }
2445
2446 /// Same as `putOne`, except does not introduce a cancelation point.
2447 ///
2448 /// For a description of cancelation and cancelation points, see `Future.cancel`.
2449 pub fn putOneUncancelable(q: *@This(), io: Io, item: Elem) QueueClosedError!void {
2450 assert(try q.putUncancelable(io, &.{item}, 1) == 1);
2451 }
2452
2453 /// Receives elements from the beginning of the queue, potentially blocking
2454 /// if there are insufficient elements currently in the queue. Returns when
2455 /// any one of the following conditions is satisfied:
2456 ///
2457 /// * At least `min` elements have been received from the queue
2458 /// * The queue is closed and contains no buffered elements
2459 /// * The current task is canceled
2460 ///
2461 /// Returns how many elements of `buffer` have been populated, if any.
2462 /// If an error is returned, no elements have been populated.
2463 ///
2464 /// If the queue is closed or the task is canceled, but some items were
2465 /// already received before the closure or cancelation, then `get` may
2466 /// return a number lower than `min`, in which case future calls are
2467 /// guaranteed to return `error.Canceled` or `error.Closed`.
2468 ///
2469 /// A return value of 0 is only possible if `min` is 0, in which case
2470 /// the call is guaranteed to fill as much of `buffer` as is possible
2471 /// *without* blocking.
2472 ///
2473 /// Asserts that `buffer.len >= min`.
2474 pub fn get(q: *@This(), io: Io, buffer: []Elem, min: usize) (QueueClosedError || Cancelable)!usize {
2475 return @divExact(try q.type_erased.get(io, @ptrCast(buffer), min * @sizeOf(Elem)), @sizeOf(Elem));
2476 }
2477
2478 /// Same as `get`, except does not introduce a cancelation point.
2479 ///
2480 /// For a description of cancelation and cancelation points, see `Future.cancel`.
2481 pub fn getUncancelable(q: *@This(), io: Io, buffer: []Elem, min: usize) QueueClosedError!usize {
2482 return @divExact(try q.type_erased.getUncancelable(io, @ptrCast(buffer), min * @sizeOf(Elem)), @sizeOf(Elem));
2483 }
2484
2485 /// Receives one element from the beginning of the queue, blocking if the queue is empty.
2486 pub fn getOne(q: *@This(), io: Io) (QueueClosedError || Cancelable)!Elem {
2487 var buf: [1]Elem = undefined;
2488 assert(try q.get(io, &buf, 1) == 1);
2489 return buf[0];
2490 }
2491
2492 /// Same as `getOne`, except does not introduce a cancelation point.
2493 ///
2494 /// For a description of cancelation and cancelation points, see `Future.cancel`.
2495 pub fn getOneUncancelable(q: *@This(), io: Io) QueueClosedError!Elem {
2496 var buf: [1]Elem = undefined;
2497 assert(try q.getUncancelable(io, &buf, 1) == 1);
2498 return buf[0];
2499 }
2500
2501 /// Returns buffer length in `Elem` units.
2502 pub fn capacity(q: *const @This()) usize {
2503 return @divExact(q.type_erased.buffer.len, @sizeOf(Elem));
2504 }
2505 };
2506}
2507
2508/// Calls `function` with `args`, such that the return value of the function is
2509/// not guaranteed to be available until `await` is called.
2510///
2511/// `function` *may* be called immediately, before `async` returns. This has
2512/// weaker guarantees than `concurrent`, making more portable and reusable.
2513///
2514/// When this function returns, it is guaranteed that `function` has already
2515/// been called and completed, or it has successfully been assigned a unit of
2516/// concurrency.
2517///
2518/// See also:
2519/// * `Group`
2520pub fn async(
2521 io: Io,
2522 function: anytype,
2523 args: std.meta.ArgsTuple(@TypeOf(function)),
2524) Future(@typeInfo(@TypeOf(function)).@"fn".return_type.?) {
2525 const Result = @typeInfo(@TypeOf(function)).@"fn".return_type.?;
2526 const Args = @TypeOf(args);
2527 const TypeErased = struct {
2528 fn start(context: *const anyopaque, result: *anyopaque) void {
2529 const args_casted: *const Args = @ptrCast(@alignCast(context));
2530 const result_casted: *Result = @ptrCast(@alignCast(result));
2531 result_casted.* = @call(.auto, function, args_casted.*);
2532 }
2533 };
2534 var future: Future(Result) = undefined;
2535 future.any_future = io.vtable.async(
2536 io.userdata,
2537 @ptrCast(&future.result),
2538 .of(Result),
2539 @ptrCast(&args),
2540 .of(Args),
2541 TypeErased.start,
2542 );
2543 return future;
2544}
2545
2546pub const ConcurrentError = error{
2547 /// May occur due to a temporary condition such as resource exhaustion, or
2548 /// to the Io implementation not supporting concurrency.
2549 ConcurrencyUnavailable,
2550};
2551
2552/// Calls `function` with `args`, such that the return value of the function is
2553/// not guaranteed to be available until `await` is called, allowing the caller
2554/// to progress while waiting for any `Io` operations.
2555///
2556/// This has stronger guarantee than `async`, placing restrictions on what kind
2557/// of `Io` implementations are supported. By calling `async` instead, one
2558/// allows, for example, stackful single-threaded blocking I/O.
2559pub fn concurrent(
2560 io: Io,
2561 function: anytype,
2562 args: std.meta.ArgsTuple(@TypeOf(function)),
2563) ConcurrentError!Future(@typeInfo(@TypeOf(function)).@"fn".return_type.?) {
2564 const Result = @typeInfo(@TypeOf(function)).@"fn".return_type.?;
2565 const Args = @TypeOf(args);
2566 const TypeErased = struct {
2567 fn start(context: *const anyopaque, result: *anyopaque) void {
2568 const args_casted: *const Args = @ptrCast(@alignCast(context));
2569 const result_casted: *Result = @ptrCast(@alignCast(result));
2570 result_casted.* = @call(.auto, function, args_casted.*);
2571 }
2572 };
2573 var future: Future(Result) = undefined;
2574 future.any_future = try io.vtable.concurrent(
2575 io.userdata,
2576 @sizeOf(Result),
2577 .of(Result),
2578 @ptrCast(&args),
2579 .of(Args),
2580 TypeErased.start,
2581 );
2582 return future;
2583}
2584
2585/// Waits until a specified amount of time has passed on `clock`.
2586///
2587/// See also:
2588/// * `Clock.Duration.sleep`
2589/// * `Clock.Timestamp.wait`
2590/// * `Timeout.sleep`
2591pub fn sleep(io: Io, duration: Duration, clock: Clock) Cancelable!void {
2592 return io.vtable.sleep(io.userdata, .{ .duration = .{
2593 .raw = duration,
2594 .clock = clock,
2595 } });
2596}
2597
2598pub const LockedStderr = struct {
2599 file_writer: *File.Writer,
2600 terminal_mode: Terminal.Mode,
2601
2602 pub fn terminal(ls: LockedStderr) Terminal {
2603 return .{
2604 .writer = &ls.file_writer.interface,
2605 .mode = ls.terminal_mode,
2606 };
2607 }
2608
2609 pub fn clear(ls: LockedStderr, buffer: []u8) Cancelable!void {
2610 const fw = ls.file_writer;
2611 std.Progress.clearWrittenWithEscapeCodes(fw) catch |err| switch (err) {
2612 error.WriteFailed => switch (fw.err.?) {
2613 error.Canceled => |e| return e,
2614 else => {},
2615 },
2616 };
2617 fw.interface.flush() catch |err| switch (err) {
2618 error.WriteFailed => switch (fw.err.?) {
2619 error.Canceled => |e| return e,
2620 else => {},
2621 },
2622 };
2623 fw.interface.buffer = buffer;
2624 }
2625};
2626
2627/// For doing application-level writes to the standard error stream.
2628/// Coordinates also with debug-level writes that are ignorant of Io interface
2629/// and implementations.
2630///
2631/// See also:
2632/// * `tryLockStderr`
2633pub fn lockStderr(io: Io, buffer: []u8, terminal_mode: ?Terminal.Mode) Cancelable!LockedStderr {
2634 const ls = try io.vtable.lockStderr(io.userdata, terminal_mode);
2635 try ls.clear(buffer);
2636 return ls;
2637}
2638
2639/// Same as `lockStderr` but non-blocking.
2640pub fn tryLockStderr(io: Io, buffer: []u8, terminal_mode: ?Terminal.Mode) Cancelable!?LockedStderr {
2641 const ls = (try io.vtable.tryLockStderr(io.userdata, terminal_mode)) orelse return null;
2642 try ls.clear(buffer);
2643 return ls;
2644}
2645
2646pub fn unlockStderr(io: Io) void {
2647 return io.vtable.unlockStderr(io.userdata);
2648}
2649
2650/// Obtains entropy from a cryptographically secure pseudo-random number
2651/// generator.
2652///
2653/// The implementation *may* store RNG state in process memory and use it to
2654/// fill `buffer`.
2655///
2656/// The randomness is seeded by `randomSecure`, or a less secure mechanism upon
2657/// failure.
2658///
2659/// Threadsafe.
2660///
2661/// See also `randomSecure`.
2662pub fn random(io: Io, buffer: []u8) void {
2663 return io.vtable.random(io.userdata, buffer);
2664}
2665
2666pub const RandomSecureError = error{EntropyUnavailable} || Cancelable;
2667
2668/// Obtains cryptographically secure entropy from outside the process.
2669///
2670/// Always makes a syscall, or otherwise avoids dependency on process memory,
2671/// in order to obtain fresh randomness. Does not rely on stored RNG state.
2672///
2673/// Does not have any fallback mechanisms; returns `error.EntropyUnavailable`
2674/// if any problems occur.
2675///
2676/// Threadsafe.
2677///
2678/// See also `random`.
2679pub fn randomSecure(io: Io, buffer: []u8) RandomSecureError!void {
2680 return io.vtable.randomSecure(io.userdata, buffer);
2681}
2682
2683test {
2684 _ = net;
2685 _ = File;
2686 _ = Dir;
2687 _ = Reader;
2688 _ = Writer;
2689 _ = Evented;
2690 _ = Threaded;
2691 _ = RwLock;
2692 _ = Semaphore;
2693 _ = @import("Io/test.zig");
2694}
2695
2696/// An implementation of `Io` which simulates a system supporting no `Io` operations.
2697///
2698/// This system has the following properties:
2699/// * Concurrency is unavailable.
2700/// * The stdio handles are pipes whose remote ends are already closed.
2701/// * The filesystem is entirely empty, including that the cwd is no longer present.
2702/// * The filesystem is full, so attempting to create entries always returns `error.NoSpaceLeft`.
2703/// * No entropy source is supported, so `randomSecure` always returns `error.EntropyUnavailable`, and `random` always returns (fills the buffer) with 0.
2704/// * No clocks are supported, so `now` and `sleep` always return `error.UnsupportedClock`.
2705/// * No network is connected, so network operations always return `error.NetworkDown`.
2706pub const failing: std.Io = .{
2707 .userdata = null,
2708 .vtable = &.{
2709 .crashHandler = noCrashHandler,
2710
2711 .async = noAsync,
2712 .concurrent = failingConcurrent,
2713 .await = unreachableAwait,
2714 .cancel = unreachableCancel,
2715
2716 .groupAsync = noGroupAsync,
2717 .groupConcurrent = failingGroupConcurrent,
2718 .groupAwait = unreachableGroupAwait,
2719 .groupCancel = unreachableGroupCancel,
2720
2721 .recancel = unreachableRecancel,
2722 .swapCancelProtection = unreachableSwapCancelProtection,
2723 .checkCancel = unreachableCheckCancel,
2724
2725 .futexWait = noFutexWait,
2726 .futexWaitUncancelable = noFutexWaitUncancelable,
2727 .futexWake = noFutexWake,
2728
2729 .operate = failingOperate,
2730 .batchAwaitAsync = unreachableBatchAwaitAsync,
2731 .batchAwaitConcurrent = unreachableBatchAwaitConcurrent,
2732 .batchCancel = unreachableBatchCancel,
2733
2734 .dirCreateDir = failingDirCreateDir,
2735 .dirCreateDirPath = failingDirCreateDirPath,
2736 .dirCreateDirPathOpen = failingDirCreateDirPathOpen,
2737 .dirOpenDir = failingDirOpenDir,
2738 .dirStat = failingDirStat,
2739 .dirStatFile = failingDirStatFile,
2740 .dirAccess = failingDirAccess,
2741 .dirCreateFile = failingDirCreateFile,
2742 .dirCreateFileAtomic = failingDirCreateFileAtomic,
2743 .dirOpenFile = failingDirOpenFile,
2744 .dirClose = unreachableDirClose,
2745 .dirRead = noDirRead,
2746 .dirRealPath = failingDirRealPath,
2747 .dirRealPathFile = failingDirRealPathFile,
2748 .dirDeleteFile = failingDirDeleteFile,
2749 .dirDeleteDir = failingDirDeleteDir,
2750 .dirRename = failingDirRename,
2751 .dirRenamePreserve = failingDirRenamePreserve,
2752 .dirSymLink = failingDirSymLink,
2753 .dirReadLink = failingDirReadLink,
2754 .dirSetOwner = failingDirSetOwner,
2755 .dirSetFileOwner = failingDirSetFileOwner,
2756 .dirSetPermissions = failingDirSetPermissions,
2757 .dirSetFilePermissions = failingDirSetFilePermissions,
2758 .dirSetTimestamps = noDirSetTimestamps,
2759 .dirHardLink = failingDirHardLink,
2760
2761 .fileStat = failingFileStat,
2762 .fileLength = failingFileLength,
2763 .fileClose = unreachableFileClose,
2764 .fileWritePositional = failingFileWritePositional,
2765 .fileWriteFileStreaming = noFileWriteFileStreaming,
2766 .fileWriteFilePositional = noFileWriteFilePositional,
2767 .fileReadPositional = failingFileReadPositional,
2768 .fileSeekBy = failingFileSeekBy,
2769 .fileSeekTo = failingFileSeekTo,
2770 .fileSync = failingFileSync,
2771 .fileIsTty = unreachableFileIsTty,
2772 .fileEnableAnsiEscapeCodes = unreachableFileEnableAnsiEscapeCodes,
2773 .fileSupportsAnsiEscapeCodes = unreachableFileSupportsAnsiEscapeCodes,
2774 .fileSetLength = failingFileSetLength,
2775 .fileSetOwner = failingFileSetOwner,
2776 .fileSetPermissions = failingFileSetPermissions,
2777 .fileSetTimestamps = noFileSetTimestamps,
2778 .fileLock = failingFileLock,
2779 .fileTryLock = failingFileTryLock,
2780 .fileUnlock = unreachableFileUnlock,
2781 .fileDowngradeLock = failingFileDowngradeLock,
2782 .fileRealPath = failingFileRealPath,
2783 .fileHardLink = failingFileHardLink,
2784
2785 .fileMemoryMapCreate = failingFileMemoryMapCreate,
2786 .fileMemoryMapDestroy = unreachableFileMemoryMapDestroy,
2787 .fileMemoryMapSetLength = unreachableFileMemoryMapSetLength,
2788 .fileMemoryMapRead = unreachableFileMemoryMapRead,
2789 .fileMemoryMapWrite = unreachableFileMemoryMapWrite,
2790
2791 .processExecutableOpen = failingProcessExecutableOpen,
2792 .processExecutablePath = failingProcessExecutablePath,
2793 .lockStderr = unreachableLockStderr,
2794 .tryLockStderr = noTryLockStderr,
2795 .unlockStderr = unreachableUnlockStderr,
2796 .processCurrentPath = failingProcessCurrentPath,
2797 .processSetCurrentDir = failingProcessSetCurrentDir,
2798 .processSetCurrentPath = failingProcessSetCurrentPath,
2799 .processReplace = failingProcessReplace,
2800 .processReplacePath = failingProcessReplacePath,
2801 .processSpawn = failingProcessSpawn,
2802 .processSpawnPath = failingProcessSpawnPath,
2803 .childWait = unreachableChildWait,
2804 .childKill = unreachableChildKill,
2805
2806 .progressParentFile = failingProgressParentFile,
2807
2808 .random = noRandom,
2809 .randomSecure = failingRandomSecure,
2810
2811 .now = noNow,
2812 .clockResolution = failingClockResolution,
2813 .sleep = noSleep,
2814
2815 .netListenIp = failingNetListenIp,
2816 .netAccept = failingNetAccept,
2817 .netBindIp = failingNetBindIp,
2818 .netConnectIp = failingNetConnectIp,
2819 .netListenUnix = failingNetListenUnix,
2820 .netConnectUnix = failingNetConnectUnix,
2821 .netSocketCreatePair = failingNetSocketCreatePair,
2822 .netWriteFile = failingNetWriteFile,
2823 .netClose = unreachableNetClose,
2824 .netShutdown = failingNetShutdown,
2825 .netInterfaceNameResolve = failingNetInterfaceNameResolve,
2826 .netInterfaceName = unreachableNetInterfaceName,
2827 .netLookup = failingNetLookup,
2828 },
2829};
2830
2831pub fn noCrashHandler(userdata: ?*anyopaque) void {
2832 _ = userdata;
2833}
2834
2835pub fn noAsync(userdata: ?*anyopaque, result: []u8, result_alignment: std.mem.Alignment, context: []const u8, context_alignment: std.mem.Alignment, start: *const fn (context: *const anyopaque, result: *anyopaque) void) ?*AnyFuture {
2836 _ = userdata;
2837 _ = result_alignment;
2838 _ = context_alignment;
2839 start(context.ptr, result.ptr);
2840 return null;
2841}
2842
2843pub fn failingConcurrent(
2844 userdata: ?*anyopaque,
2845 result_len: usize,
2846 result_alignment: std.mem.Alignment,
2847 context: []const u8,
2848 context_alignment: std.mem.Alignment,
2849 start: *const fn (context: *const anyopaque, result: *anyopaque) void,
2850) ConcurrentError!*AnyFuture {
2851 _ = userdata;
2852 _ = result_len;
2853 _ = result_alignment;
2854 _ = context;
2855 _ = context_alignment;
2856 _ = start;
2857 return error.ConcurrencyUnavailable;
2858}
2859
2860pub fn unreachableAwait(
2861 userdata: ?*anyopaque,
2862 any_future: *AnyFuture,
2863 result: []u8,
2864 result_alignment: std.mem.Alignment,
2865) void {
2866 _ = userdata;
2867 _ = any_future;
2868 _ = result;
2869 _ = result_alignment;
2870 unreachable;
2871}
2872
2873pub fn unreachableCancel(
2874 userdata: ?*anyopaque,
2875 any_future: *AnyFuture,
2876 result: []u8,
2877 result_alignment: std.mem.Alignment,
2878) void {
2879 _ = userdata;
2880 _ = any_future;
2881 _ = result;
2882 _ = result_alignment;
2883 unreachable;
2884}
2885
2886pub fn noGroupAsync(
2887 userdata: ?*anyopaque,
2888 group: *Group,
2889 context: []const u8,
2890 context_alignment: std.mem.Alignment,
2891 start: *const fn (context: *const anyopaque) void,
2892) void {
2893 _ = userdata;
2894 _ = group;
2895 _ = context_alignment;
2896 start(context.ptr);
2897}
2898
2899pub fn failingGroupConcurrent(
2900 userdata: ?*anyopaque,
2901 group: *Group,
2902 context: []const u8,
2903 context_alignment: std.mem.Alignment,
2904 start: *const fn (context: *const anyopaque) void,
2905) ConcurrentError!void {
2906 _ = userdata;
2907 _ = group;
2908 _ = context;
2909 _ = context_alignment;
2910 _ = start;
2911 return error.ConcurrencyUnavailable;
2912}
2913
2914pub fn unreachableGroupAwait(userdata: ?*anyopaque, group: *Group, token: *anyopaque) Cancelable!void {
2915 _ = userdata;
2916 _ = group;
2917 _ = token;
2918 unreachable;
2919}
2920
2921pub fn unreachableGroupCancel(userdata: ?*anyopaque, group: *Group, token: *anyopaque) void {
2922 _ = userdata;
2923 _ = group;
2924 _ = token;
2925 unreachable;
2926}
2927
2928pub fn unreachableRecancel(userdata: ?*anyopaque) void {
2929 _ = userdata;
2930 unreachable;
2931}
2932
2933pub fn unreachableSwapCancelProtection(userdata: ?*anyopaque, new: CancelProtection) CancelProtection {
2934 _ = userdata;
2935 _ = new;
2936 unreachable;
2937}
2938
2939pub fn unreachableCheckCancel(userdata: ?*anyopaque) Cancelable!void {
2940 _ = userdata;
2941 unreachable;
2942}
2943
2944pub fn noFutexWait(userdata: ?*anyopaque, ptr: *const u32, expected: u32, timeout: Timeout) Cancelable!void {
2945 _ = userdata;
2946 std.debug.assert(ptr.* == expected or timeout != .none);
2947}
2948
2949pub fn noFutexWaitUncancelable(userdata: ?*anyopaque, ptr: *const u32, expected: u32) void {
2950 _ = userdata;
2951 std.debug.assert(ptr.* == expected);
2952}
2953
2954pub fn noFutexWake(userdata: ?*anyopaque, ptr: *const u32, max_waiters: u32) void {
2955 _ = userdata;
2956 _ = ptr;
2957 _ = max_waiters;
2958 // no-op
2959}
2960
2961pub fn failingOperate(userdata: ?*anyopaque, operation: Operation) Cancelable!Operation.Result {
2962 _ = userdata;
2963 return switch (operation) {
2964 .file_read_streaming => .{ .file_read_streaming = error.InputOutput },
2965 .file_write_streaming => .{ .file_write_streaming = error.InputOutput },
2966 .device_io_control => unreachable,
2967 .net_receive => .{ .net_receive = .{ error.NetworkDown, 0 } },
2968 .net_send => .{ .net_send = .{ error.NetworkDown, 0 } },
2969 .net_read => .{ .net_read = error.NetworkDown },
2970 .net_write => .{ .net_write = error.NetworkDown },
2971 };
2972}
2973
2974pub fn unreachableBatchAwaitAsync(userdata: ?*anyopaque, b: *Batch) Cancelable!void {
2975 _ = userdata;
2976 _ = b;
2977 unreachable;
2978}
2979
2980pub fn unreachableBatchAwaitConcurrent(userdata: ?*anyopaque, b: *Batch, timeout: Timeout) Batch.AwaitConcurrentError!void {
2981 _ = userdata;
2982 _ = b;
2983 _ = timeout;
2984 unreachable;
2985}
2986
2987pub fn unreachableBatchCancel(userdata: ?*anyopaque, b: *Batch) void {
2988 _ = userdata;
2989 _ = b;
2990 unreachable;
2991}
2992
2993pub fn failingDirCreateDir(userdata: ?*anyopaque, dir: Dir, sub_path: []const u8, permissions: Dir.Permissions) Dir.CreateDirError!void {
2994 _ = userdata;
2995 _ = dir;
2996 _ = sub_path;
2997 _ = permissions;
2998 return error.NoSpaceLeft;
2999}
3000
3001pub fn failingDirCreateDirPath(userdata: ?*anyopaque, dir: Dir, sub_path: []const u8, permissions: Dir.Permissions) Dir.CreateDirPathError!Dir.CreatePathStatus {
3002 _ = userdata;
3003 _ = dir;
3004 _ = sub_path;
3005 _ = permissions;
3006 return error.NoSpaceLeft;
3007}
3008
3009pub fn failingDirCreateDirPathOpen(userdata: ?*anyopaque, dir: Dir, sub_path: []const u8, permissions: Dir.Permissions, options: Dir.OpenOptions) Dir.CreateDirPathOpenError!Dir {
3010 _ = userdata;
3011 _ = dir;
3012 _ = sub_path;
3013 _ = permissions;
3014 _ = options;
3015 return error.NoSpaceLeft;
3016}
3017
3018pub fn failingDirOpenDir(userdata: ?*anyopaque, dir: Dir, sub_path: []const u8, options: Dir.OpenOptions) Dir.OpenError!Dir {
3019 _ = userdata;
3020 _ = dir;
3021 _ = sub_path;
3022 _ = options;
3023 return error.FileNotFound;
3024}
3025
3026pub fn failingDirStat(userdata: ?*anyopaque, dir: Dir) Dir.StatError!Dir.Stat {
3027 _ = userdata;
3028 _ = dir;
3029 return error.Streaming;
3030}
3031
3032pub fn failingDirStatFile(userdata: ?*anyopaque, dir: Dir, sub_path: []const u8, options: Dir.StatFileOptions) Dir.StatFileError!File.Stat {
3033 _ = userdata;
3034 _ = dir;
3035 _ = sub_path;
3036 _ = options;
3037 return error.FileNotFound;
3038}
3039
3040pub fn failingDirAccess(userdata: ?*anyopaque, dir: Dir, sub_path: []const u8, options: Dir.AccessOptions) Dir.AccessError!void {
3041 _ = userdata;
3042 _ = dir;
3043 _ = sub_path;
3044 _ = options;
3045 return error.FileNotFound;
3046}
3047
3048pub fn failingDirCreateFile(userdata: ?*anyopaque, dir: Dir, sub_path: []const u8, options: Dir.CreateFileOptions) File.OpenError!File {
3049 _ = userdata;
3050 _ = dir;
3051 _ = sub_path;
3052 _ = options;
3053 return error.NoSpaceLeft;
3054}
3055
3056pub fn failingDirCreateFileAtomic(userdata: ?*anyopaque, dir: Dir, sub_path: []const u8, options: Dir.CreateFileAtomicOptions) Dir.CreateFileAtomicError!File.Atomic {
3057 _ = userdata;
3058 _ = dir;
3059 _ = sub_path;
3060 _ = options;
3061 return error.NoSpaceLeft;
3062}
3063
3064pub fn failingDirOpenFile(userdata: ?*anyopaque, dir: Dir, sub_path: []const u8, flags: Dir.OpenFileOptions) File.OpenError!File {
3065 _ = userdata;
3066 _ = dir;
3067 _ = sub_path;
3068 _ = flags;
3069 return error.FileNotFound;
3070}
3071
3072pub fn unreachableDirClose(userdata: ?*anyopaque, dirs: []const Dir) void {
3073 _ = userdata;
3074 _ = dirs;
3075 unreachable;
3076}
3077
3078pub fn noDirRead(userdata: ?*anyopaque, dir_reader: *Dir.Reader, buffer: []Dir.Entry) Dir.Reader.Error!usize {
3079 _ = userdata;
3080 _ = dir_reader;
3081 _ = buffer;
3082 return 0;
3083}
3084
3085pub fn failingDirRealPath(userdata: ?*anyopaque, dir: Dir, out_buffer: []u8) Dir.RealPathError!usize {
3086 _ = userdata;
3087 _ = dir;
3088 _ = out_buffer;
3089 return error.FileNotFound;
3090}
3091
3092pub fn failingDirRealPathFile(userdata: ?*anyopaque, dir: Dir, path_name: []const u8, out_buffer: []u8) Dir.RealPathFileError!usize {
3093 _ = userdata;
3094 _ = dir;
3095 _ = path_name;
3096 _ = out_buffer;
3097 return error.FileNotFound;
3098}
3099
3100pub fn failingDirDeleteFile(userdata: ?*anyopaque, dir: Dir, sub_path: []const u8) Dir.DeleteFileError!void {
3101 _ = userdata;
3102 _ = dir;
3103 _ = sub_path;
3104 return error.FileNotFound;
3105}
3106
3107pub fn failingDirDeleteDir(userdata: ?*anyopaque, dir: Dir, sub_path: []const u8) Dir.DeleteDirError!void {
3108 _ = userdata;
3109 _ = dir;
3110 _ = sub_path;
3111 return error.FileNotFound;
3112}
3113
3114pub fn failingDirRename(userdata: ?*anyopaque, old_dir: Dir, old_sub_path: []const u8, new_dir: Dir, new_sub_path: []const u8) Dir.RenameError!void {
3115 _ = userdata;
3116 _ = old_dir;
3117 _ = old_sub_path;
3118 _ = new_dir;
3119 _ = new_sub_path;
3120 return error.FileNotFound;
3121}
3122
3123pub fn failingDirRenamePreserve(userdata: ?*anyopaque, old_dir: Dir, old_sub_path: []const u8, new_dir: Dir, new_sub_path: []const u8) Dir.RenamePreserveError!void {
3124 _ = userdata;
3125 _ = old_dir;
3126 _ = old_sub_path;
3127 _ = new_dir;
3128 _ = new_sub_path;
3129 return error.FileNotFound;
3130}
3131
3132pub fn failingDirSymLink(userdata: ?*anyopaque, dir: Dir, target_path: []const u8, sym_link_path: []const u8, flags: Dir.SymLinkFlags) Dir.SymLinkError!void {
3133 _ = userdata;
3134 _ = dir;
3135 _ = target_path;
3136 _ = sym_link_path;
3137 _ = flags;
3138 return error.FileNotFound;
3139}
3140
3141pub fn failingDirReadLink(userdata: ?*anyopaque, dir: Dir, sub_path: []const u8, buffer: []u8) Dir.ReadLinkError!usize {
3142 _ = userdata;
3143 _ = dir;
3144 _ = sub_path;
3145 _ = buffer;
3146 return error.FileNotFound;
3147}
3148
3149pub fn failingDirSetOwner(userdata: ?*anyopaque, dir: Dir, owner: ?File.Uid, group: ?File.Gid) Dir.SetOwnerError!void {
3150 _ = userdata;
3151 _ = dir;
3152 _ = owner;
3153 _ = group;
3154 return error.FileNotFound;
3155}
3156
3157pub fn failingDirSetFileOwner(userdata: ?*anyopaque, dir: std.Io.Dir, sub_path: []const u8, owner: ?File.Uid, group: ?File.Gid, options: Dir.SetFileOwnerOptions) Dir.SetFileOwnerError!void {
3158 _ = userdata;
3159 _ = dir;
3160 _ = sub_path;
3161 _ = owner;
3162 _ = group;
3163 _ = options;
3164 return error.FileNotFound;
3165}
3166
3167pub fn failingDirSetPermissions(userdata: ?*anyopaque, dir: Dir, permissions: Dir.Permissions) Dir.SetPermissionsError!void {
3168 _ = userdata;
3169 _ = dir;
3170 _ = permissions;
3171 return error.FileNotFound;
3172}
3173
3174pub fn failingDirSetFilePermissions(userdata: ?*anyopaque, dir: Dir, sub_path: []const u8, permissions: File.Permissions, options: Dir.SetFilePermissionsOptions) Dir.SetFilePermissionsError!void {
3175 _ = userdata;
3176 _ = dir;
3177 _ = sub_path;
3178 _ = permissions;
3179 _ = options;
3180 return error.FileNotFound;
3181}
3182
3183pub fn noDirSetTimestamps(userdata: ?*anyopaque, dir: Dir, sub_path: []const u8, options: Dir.SetTimestampsOptions) Dir.SetTimestampsError!void {
3184 _ = userdata;
3185 _ = dir;
3186 _ = sub_path;
3187 _ = options;
3188 // no-op
3189}
3190
3191pub fn failingDirHardLink(userdata: ?*anyopaque, old_dir: Dir, old_sub_path: []const u8, new_dir: Dir, new_sub_path: []const u8, options: Dir.HardLinkOptions) Dir.HardLinkError!void {
3192 _ = userdata;
3193 _ = old_dir;
3194 _ = old_sub_path;
3195 _ = new_dir;
3196 _ = new_sub_path;
3197 _ = options;
3198 return error.FileNotFound;
3199}
3200
3201pub fn failingFileStat(userdata: ?*anyopaque, file: File) File.StatError!File.Stat {
3202 _ = userdata;
3203 _ = file;
3204 return error.Streaming;
3205}
3206
3207pub fn failingFileLength(userdata: ?*anyopaque, file: File) File.LengthError!u64 {
3208 _ = userdata;
3209 _ = file;
3210 return error.Streaming;
3211}
3212
3213pub fn unreachableFileClose(userdata: ?*anyopaque, files: []const File) void {
3214 _ = userdata;
3215 _ = files;
3216 unreachable;
3217}
3218
3219pub fn failingFileWritePositional(userdata: ?*anyopaque, file: File, header: []const u8, data: []const []const u8, splat: usize, offset: u64) File.WritePositionalError!usize {
3220 _ = userdata;
3221 _ = file;
3222 _ = header;
3223 _ = offset;
3224 for (data[0 .. data.len - 1]) |item| {
3225 if (item.len > 0) return error.BrokenPipe;
3226 }
3227 if (data[data.len - 1].len != 0 and splat != 0) return error.BrokenPipe;
3228 return 0;
3229}
3230
3231pub fn noFileWriteFileStreaming(userdata: ?*anyopaque, file: File, header: []const u8, file_reader: *Io.File.Reader, limit: Io.Limit) File.Writer.WriteFileError!usize {
3232 _ = userdata;
3233 _ = file;
3234 _ = header;
3235 _ = file_reader;
3236 _ = limit;
3237 return error.Unimplemented;
3238}
3239
3240pub fn noFileWriteFilePositional(userdata: ?*anyopaque, file: File, header: []const u8, file_reader: *Io.File.Reader, limit: Io.Limit, offset: u64) File.WriteFilePositionalError!usize {
3241 _ = userdata;
3242 _ = file;
3243 _ = header;
3244 _ = file_reader;
3245 _ = limit;
3246 _ = offset;
3247 return error.Unimplemented;
3248}
3249
3250pub fn failingFileReadPositional(userdata: ?*anyopaque, file: File, data: []const []u8, offset: u64) File.ReadPositionalError!usize {
3251 _ = userdata;
3252 _ = file;
3253 _ = offset;
3254 for (data) |item| {
3255 if (item.len > 0) return error.InputOutput;
3256 }
3257 return 0;
3258}
3259
3260pub fn failingFileSeekBy(userdata: ?*anyopaque, file: File, relative_offset: i64) File.SeekError!void {
3261 _ = userdata;
3262 _ = file;
3263 _ = relative_offset;
3264 return error.Unseekable;
3265}
3266
3267pub fn failingFileSeekTo(userdata: ?*anyopaque, file: File, absolute_offset: u64) File.SeekError!void {
3268 _ = userdata;
3269 _ = file;
3270 _ = absolute_offset;
3271 return error.Unseekable;
3272}
3273
3274pub fn failingFileSync(userdata: ?*anyopaque, file: File) File.SyncError!void {
3275 _ = userdata;
3276 _ = file;
3277 return error.NoSpaceLeft;
3278}
3279
3280pub fn unreachableFileIsTty(userdata: ?*anyopaque, file: File) Cancelable!bool {
3281 _ = userdata;
3282 _ = file;
3283 unreachable;
3284}
3285
3286pub fn unreachableFileEnableAnsiEscapeCodes(userdata: ?*anyopaque, file: File) File.EnableAnsiEscapeCodesError!void {
3287 _ = userdata;
3288 _ = file;
3289 unreachable;
3290}
3291
3292pub fn unreachableFileSupportsAnsiEscapeCodes(userdata: ?*anyopaque, file: File) Cancelable!bool {
3293 _ = userdata;
3294 _ = file;
3295 unreachable;
3296}
3297
3298pub fn failingFileSetLength(userdata: ?*anyopaque, file: File, length: u64) File.SetLengthError!void {
3299 _ = userdata;
3300 _ = file;
3301 _ = length;
3302 return error.NonResizable;
3303}
3304
3305pub fn failingFileSetOwner(userdata: ?*anyopaque, file: File, owner: ?File.Uid, group: ?File.Gid) File.SetOwnerError!void {
3306 _ = userdata;
3307 _ = file;
3308 _ = owner;
3309 _ = group;
3310 return error.FileNotFound;
3311}
3312
3313pub fn failingFileSetPermissions(userdata: ?*anyopaque, file: File, permissions: File.Permissions) File.SetPermissionsError!void {
3314 _ = userdata;
3315 _ = file;
3316 _ = permissions;
3317 return error.FileNotFound;
3318}
3319
3320pub fn noFileSetTimestamps(userdata: ?*anyopaque, file: File, options: File.SetTimestampsOptions) File.SetTimestampsError!void {
3321 _ = userdata;
3322 _ = file;
3323 _ = options;
3324 // no-op
3325}
3326
3327pub fn failingFileLock(userdata: ?*anyopaque, file: File, lock: File.Lock) File.LockError!void {
3328 _ = userdata;
3329 _ = file;
3330 _ = lock;
3331 return error.FileLocksUnsupported;
3332}
3333
3334pub fn failingFileTryLock(userdata: ?*anyopaque, file: File, lock: File.Lock) File.LockError!bool {
3335 _ = userdata;
3336 _ = file;
3337 _ = lock;
3338 return error.FileLocksUnsupported;
3339}
3340
3341pub fn unreachableFileUnlock(userdata: ?*anyopaque, file: File) void {
3342 _ = userdata;
3343 _ = file;
3344 unreachable;
3345}
3346
3347pub fn failingFileDowngradeLock(userdata: ?*anyopaque, file: File) File.DowngradeLockError!void {
3348 _ = userdata;
3349 _ = file;
3350 // no-op
3351}
3352
3353pub fn failingFileRealPath(userdata: ?*anyopaque, file: File, out_buffer: []u8) File.RealPathError!usize {
3354 _ = userdata;
3355 _ = file;
3356 _ = out_buffer;
3357 return error.FileNotFound;
3358}
3359
3360pub fn failingFileHardLink(userdata: ?*anyopaque, file: File, new_dir: Dir, new_sub_path: []const u8, options: File.HardLinkOptions) File.HardLinkError!void {
3361 _ = userdata;
3362 _ = file;
3363 _ = new_dir;
3364 _ = new_sub_path;
3365 _ = options;
3366 return error.FileNotFound;
3367}
3368
3369pub fn failingFileMemoryMapCreate(userdata: ?*anyopaque, file: File, options: File.MemoryMap.CreateOptions) File.MemoryMap.CreateError!File.MemoryMap {
3370 _ = userdata;
3371 _ = file;
3372 _ = options;
3373 return error.AccessDenied;
3374}
3375
3376pub fn unreachableFileMemoryMapDestroy(userdata: ?*anyopaque, mm: *File.MemoryMap) void {
3377 _ = userdata;
3378 _ = mm;
3379 unreachable;
3380}
3381
3382pub fn unreachableFileMemoryMapSetLength(userdata: ?*anyopaque, mm: *File.MemoryMap, new_len: usize) File.MemoryMap.SetLengthError!void {
3383 _ = userdata;
3384 _ = mm;
3385 _ = new_len;
3386 unreachable;
3387}
3388
3389pub fn unreachableFileMemoryMapRead(userdata: ?*anyopaque, mm: *File.MemoryMap) File.ReadPositionalError!void {
3390 _ = userdata;
3391 _ = mm;
3392 unreachable;
3393}
3394
3395pub fn unreachableFileMemoryMapWrite(userdata: ?*anyopaque, mm: *File.MemoryMap) File.WritePositionalError!void {
3396 _ = userdata;
3397 _ = mm;
3398 unreachable;
3399}
3400
3401pub fn failingProcessExecutableOpen(userdata: ?*anyopaque, flags: Dir.OpenFileOptions) std.process.OpenExecutableError!File {
3402 _ = userdata;
3403 _ = flags;
3404 return error.FileNotFound;
3405}
3406
3407pub fn failingProcessExecutablePath(userdata: ?*anyopaque, buffer: []u8) std.process.ExecutablePathError!usize {
3408 _ = userdata;
3409 _ = buffer;
3410 return error.FileNotFound;
3411}
3412
3413pub fn unreachableLockStderr(userdata: ?*anyopaque, terminal_mode: ?Terminal.Mode) Cancelable!LockedStderr {
3414 _ = userdata;
3415 _ = terminal_mode;
3416 unreachable;
3417}
3418
3419pub fn noTryLockStderr(userdata: ?*anyopaque, terminal_mode: ?Terminal.Mode) Cancelable!?LockedStderr {
3420 _ = userdata;
3421 _ = terminal_mode;
3422 return null;
3423}
3424
3425pub fn unreachableUnlockStderr(userdata: ?*anyopaque) void {
3426 _ = userdata;
3427 unreachable;
3428}
3429
3430pub fn failingProcessCurrentPath(userdata: ?*anyopaque, buffer: []u8) std.process.CurrentPathError!usize {
3431 _ = userdata;
3432 _ = buffer;
3433 return error.CurrentDirUnlinked;
3434}
3435
3436pub fn failingProcessSetCurrentDir(userdata: ?*anyopaque, dir: Dir) std.process.SetCurrentDirError!void {
3437 _ = userdata;
3438 _ = dir;
3439 return error.FileNotFound;
3440}
3441
3442pub fn failingProcessSetCurrentPath(userdata: ?*anyopaque, path: []const u8) std.process.SetCurrentPathError!void {
3443 _ = userdata;
3444 _ = path;
3445 return error.FileNotFound;
3446}
3447
3448pub fn failingProcessReplace(userdata: ?*anyopaque, options: std.process.ReplaceOptions) std.process.ReplaceError {
3449 _ = userdata;
3450 _ = options;
3451 return error.OperationUnsupported;
3452}
3453
3454pub fn failingProcessReplacePath(userdata: ?*anyopaque, dir: Dir, options: std.process.ReplaceOptions) std.process.ReplaceError {
3455 _ = userdata;
3456 _ = dir;
3457 _ = options;
3458 return error.OperationUnsupported;
3459}
3460
3461pub fn failingProcessSpawn(userdata: ?*anyopaque, options: std.process.SpawnOptions) std.process.SpawnError!std.process.Child {
3462 _ = userdata;
3463 _ = options;
3464 return error.OperationUnsupported;
3465}
3466
3467pub fn failingProcessSpawnPath(userdata: ?*anyopaque, dir: Dir, options: std.process.SpawnOptions) std.process.SpawnError!std.process.Child {
3468 _ = userdata;
3469 _ = dir;
3470 _ = options;
3471 return error.OperationUnsupported;
3472}
3473
3474pub fn unreachableChildWait(userdata: ?*anyopaque, child: *std.process.Child) std.process.Child.WaitError!std.process.Child.Term {
3475 _ = userdata;
3476 _ = child;
3477 unreachable;
3478}
3479
3480pub fn unreachableChildKill(userdata: ?*anyopaque, child: *std.process.Child) void {
3481 _ = userdata;
3482 _ = child;
3483 unreachable;
3484}
3485
3486pub fn failingProgressParentFile(userdata: ?*anyopaque) std.Progress.ParentFileError!File {
3487 _ = userdata;
3488 return error.UnsupportedOperation;
3489}
3490
3491pub fn noRandom(userdata: ?*anyopaque, buffer: []u8) void {
3492 _ = userdata;
3493 @memset(buffer, 0);
3494}
3495
3496pub fn failingRandomSecure(userdata: ?*anyopaque, buffer: []u8) RandomSecureError!void {
3497 _ = userdata;
3498 _ = buffer;
3499 return error.EntropyUnavailable;
3500}
3501
3502pub fn noNow(userdata: ?*anyopaque, clock: Clock) Timestamp {
3503 _ = userdata;
3504 _ = clock;
3505 return .zero;
3506}
3507
3508pub fn failingClockResolution(userdata: ?*anyopaque, clock: Clock) Clock.ResolutionError!Duration {
3509 _ = userdata;
3510 _ = clock;
3511 return error.ClockUnavailable;
3512}
3513
3514pub fn noSleep(userdata: ?*anyopaque, clock: Timeout) Cancelable!void {
3515 _ = userdata;
3516 _ = clock;
3517}
3518
3519pub fn failingNetListenIp(userdata: ?*anyopaque, address: *const net.IpAddress, options: net.IpAddress.ListenOptions) net.IpAddress.ListenError!net.Socket {
3520 _ = userdata;
3521 _ = address;
3522 _ = options;
3523 return error.NetworkDown;
3524}
3525
3526pub fn failingNetAccept(userdata: ?*anyopaque, listen_fd: net.Socket.Handle, options: net.Server.AcceptOptions) net.Server.AcceptError!net.Socket {
3527 _ = userdata;
3528 _ = listen_fd;
3529 _ = options;
3530 return error.NetworkDown;
3531}
3532
3533pub fn failingNetBindIp(userdata: ?*anyopaque, address: *const net.IpAddress, options: net.IpAddress.BindOptions) net.IpAddress.BindError!net.Socket {
3534 _ = userdata;
3535 _ = address;
3536 _ = options;
3537 return error.NetworkDown;
3538}
3539
3540pub fn failingNetConnectIp(userdata: ?*anyopaque, address: *const net.IpAddress, options: net.IpAddress.ConnectOptions) net.IpAddress.ConnectError!net.Socket {
3541 _ = userdata;
3542 _ = address;
3543 _ = options;
3544 return error.NetworkDown;
3545}
3546
3547pub fn failingNetListenUnix(userdata: ?*anyopaque, address: *const net.UnixAddress, options: net.UnixAddress.ListenOptions) net.UnixAddress.ListenError!net.Socket.Handle {
3548 _ = userdata;
3549 _ = address;
3550 _ = options;
3551 return error.NetworkDown;
3552}
3553
3554pub fn failingNetConnectUnix(userdata: ?*anyopaque, address: *const net.UnixAddress) net.UnixAddress.ConnectError!net.Socket.Handle {
3555 _ = userdata;
3556 _ = address;
3557 return error.NetworkDown;
3558}
3559
3560pub fn failingNetSocketCreatePair(userdata: ?*anyopaque, options: net.Socket.CreatePairOptions) net.Socket.CreatePairError![2]net.Socket {
3561 _ = userdata;
3562 _ = options;
3563 return error.OperationUnsupported;
3564}
3565
3566pub fn failingNetWriteFile(userdata: ?*anyopaque, handle: net.Socket.Handle, header: []const u8, file_reader: *Io.File.Reader, limit: Io.Limit) net.Stream.Writer.WriteFileError!usize {
3567 _ = userdata;
3568 _ = handle;
3569 _ = header;
3570 _ = file_reader;
3571 _ = limit;
3572 return error.NetworkDown;
3573}
3574
3575pub fn unreachableNetClose(userdata: ?*anyopaque, sockets: []const net.Socket) void {
3576 _ = userdata;
3577 _ = sockets;
3578 unreachable;
3579}
3580
3581pub fn failingNetShutdown(userdata: ?*anyopaque, handle: net.Socket.Handle, how: net.ShutdownHow) net.ShutdownError!void {
3582 _ = userdata;
3583 _ = handle;
3584 _ = how;
3585 return error.NetworkDown;
3586}
3587
3588pub fn failingNetInterfaceNameResolve(userdata: ?*anyopaque, name: *const net.Interface.Name) net.Interface.Name.ResolveError!net.Interface {
3589 _ = userdata;
3590 _ = name;
3591 return error.InterfaceNotFound;
3592}
3593
3594pub fn unreachableNetInterfaceName(userdata: ?*anyopaque, interface: net.Interface) net.Interface.NameError!net.Interface.Name {
3595 _ = userdata;
3596 _ = interface;
3597 unreachable;
3598}
3599
3600pub fn failingNetLookup(userdata: ?*anyopaque, host_name: net.HostName, resolved: *Queue(net.HostName.LookupResult), options: net.HostName.LookupOptions) net.HostName.LookupError!void {
3601 _ = userdata;
3602 _ = host_name;
3603 _ = resolved;
3604 _ = options;
3605 return error.NetworkDown;
3606}
3607
3608test failing {
3609 const f: Io = .failing;
3610 // file stuff
3611 try std.testing.expectError(error.NoSpaceLeft, Dir.createDir(.cwd(), f, "test", .default_dir));
3612 try std.testing.expectError(error.NoSpaceLeft, Dir.createFile(.cwd(), f, "test", .{}));
3613 try std.testing.expectError(error.FileNotFound, Dir.openDir(.cwd(), f, "test", .{}));
3614 try std.testing.expectError(error.FileNotFound, Dir.openFile(.cwd(), f, "test", .{}));
3615 try File.writeStreamingAll(.stdout(), f, &.{});
3616 try std.testing.expectError(error.AccessDenied, File.MemoryMap.create(f, .stdout(), .{ .len = 0 }));
3617 // async stuff
3618 const closure = struct {
3619 var foo: usize = 0;
3620 fn doOp() void {
3621 foo = 4;
3622 }
3623 };
3624 var future = f.async(closure.doOp, .{});
3625 _ = future.await(f);
3626 try std.testing.expect(closure.foo == 4);
3627 // random stuff
3628 var buffer: [1]u8 = undefined;
3629 f.random(&buffer);
3630 try std.testing.expect(buffer[0] == 0);
3631}