#!/usr/bin/env dub
/+ dub.sdl:
name "gcd_source_read_pipe"
platforms "osx"
targetPath "build"
+/
/**
* GCD — `DISPATCH_SOURCE_TYPE_READ` is `EVFILT_READ` with the loop hidden.
*
* A dispatch source is libdispatch's whole event-loop surface: you never call
* `kqueue()` or `kevent()`, you attach a handler to a source and the kernel
* delivers the event straight to a workqueue thread. `_dispatch_source_type_read`
* (`src/event/event.c`) is literally `{ .dst_filter = EVFILT_READ, .dst_flags =
* EV_UDATA_SPECIFIC|EV_DISPATCH|EV_VANISHED }`, so the observable behaviour is
* kqueue's: `dispatch_source_get_data()` returns the same byte count kqueue puts
* in `kevent.data`, the registration auto-disables itself while the handler runs
* (`EV_DISPATCH`), and a closed writer surfaces as a wakeup with zero bytes.
*
* The program drives a pipe through three states — readable, drained, EOF —
* sequencing each with a semaphore so the output is deterministic, and asserts
* the byte counts and the cancel-handler ordering.
*
* Companion to the GCD deep-dive:
* see docs/research/async-io/gcd/index.md § "Dispatch sources: a kqueue vocabulary".
*
* Run with: `dub run --single source-read-pipe.d`
*
* Portability: macOS only (`platforms "osx"`).
*/
module (module) gcd_source_read_pipeGCD — DISPATCH_SOURCE_TYPE_READ is EVFILT_READ with the loop hidden.
A dispatch source is libdispatch's whole event-loop surface: you never call
kqueue() or kevent(), you attach a handler to a source and the kernel
delivers the event straight to a workqueue thread. _dispatch_source_type_read
(src/event/event.c) is literally { .dst_filter = EVFILT_READ, .dst_flags =
EV_UDATA_SPECIFIC|EV_DISPATCH|EV_VANISHED }, so the observable behaviour is
kqueue's: dispatch_source_get_data() returns the same byte count kqueue puts
in kevent.data, the registration auto-disables itself while the handler runs
(EV_DISPATCH), and a closed writer surfaces as a wakeup with zero bytes.
The program drives a pipe through three states — readable, drained, EOF —
sequencing each with a semaphore so the output is deterministic, and asserts
the byte counts and the cancel-handler ordering.
Companion to the GCD deep-dive:
see docs/research/async-io/gcd/index.md § "Dispatch sources: a kqueue vocabulary".
Run with: dub run --single source-read-pipe.d
Portability
macOS only (platforms "osx").
gcd_source_read_pipe;
import (package) corecore.(module) core.atomicThe atomic module provides basic support for lock-free
concurrent programming.
Use the -preview=nosharedaccess compiler flag to detect
unsafe individual read or write operations on shared data.
Source
core/atomic.d
Examples
int y = 2;
shared int x = y; // OK
//x++; // read modify write error
x.atomicOp!"+="(1); // OK
//y = x; // read error with preview flag
y = x.atomicLoad(); // OK
assert(y == 3);
//x = 5; // write error with preview flag
x.atomicStore(5); // OK
assert(x.atomicLoad() == 5);
atomic : (alias template) gcd_source_read_pipe.atomicLoad = core.atomic.atomicLoad(MemoryOrder ms = MemoryOrder.seq, T)(auto ref return scope const T val) if (!is(T == shared(U), U) && !is(T == shared(inout(U)), U) && !is(T == shared(const(U)), U))Loads 'val' from memory and returns it. The memory barrier specified
by 'ms' is applied to the operation, which is fully sequenced by
default. Valid memory orders are MemoryOrder.raw, MemoryOrder.acq,
and MemoryOrder.seq.
atomicLoad, (alias template) gcd_source_read_pipe.atomicOp = core.atomic.atomicOp(string op, T, V1)(ref shared T val, V1 mod) if (__traits(compiles, mixin("*cast(T*)&val" ~ op ~ "mod")))Performs the binary operation 'op' on val using 'mod' as the modifier.
atomicOp, (alias template) gcd_source_read_pipe.atomicStore = core.atomic.atomicStore(MemoryOrder ms = MemoryOrder.seq, T, V)(ref T val, V newval) if (!is(T == shared) && !is(V == shared))Writes 'newval' into 'val'. The memory barrier specified by 'ms' is
applied to the operation, which is fully sequenced by default.
Valid memory orders are MemoryOrder.raw, MemoryOrder.rel, and
MemoryOrder.seq.
atomicStore;
import (package) corecore.(package) core.stdcstdc.(module) core.stdc.stdintD header file for C99.
pubs.opengroup.org/onlinepubs/009695399/basedefs/stdint.h.html, stdint.h
Source
core/stdc/stdint.d
stdint : uintptr_t;
import (package) corecore.(package) core.syssys.(package) core.sys.posixposix.(module) core.sys.posix.unistdD header file for POSIX.
unistd : (alias) gcd_source_read_pipe.close = int core.sys.posix.unistd.close(int) nothrow @nogc @trustedclose, (alias) gcd_source_read_pipe.pipe = int core.sys.posix.unistd.pipe(ref int[2]) nothrow @nogc @trustedpipe, (alias) gcd_source_read_pipe.read = long core.sys.posix.unistd.read(int, void*, ulong) nothrow @nogcread, (alias) gcd_source_read_pipe.write = long core.sys.posix.unistd.write(int, scope const(void*), ulong) nothrow @nogcwrite;
import (package) stdstd.(module) std.stdioCategory Symbols File handles _popen File isFileHandle openNetwork stderr stdin stdout Reading chunks lines readf readfln readln Writing toFile write writef writefln writeln Misc KeepTerminator LockType StdioException
Standard I/O functions that extend core.stdc.stdio. core.stdc.stdio
is publically imported when importing std.stdio.
There are three layers of I/O:
The lowest layer is the operating system layer. The two main schemes are Windows and Posix.
C's stdio.h which unifies the two operating system schemes.
std.stdio, this module, unifies the various stdio.h implementations into
a high level package for D programs.
Source
std/stdio.d
stdio : (alias template) gcd_source_read_pipe.writefln = std.stdio.writefln(alias fmt, A...)(A args) if (isSomeString!(typeof(fmt)))Equivalent to writef(fmt, args, '\n').
writefln, (alias template) gcd_source_read_pipe.writeln = std.stdio.writeln(T...)(T args)Equivalent to write(args, '\n'). Calling writeln without
arguments is valid and just prints a newline to the standard
output.
Example
Reads stdin and writes it to stdout with an argument
counter.
import std.stdio;
void main()
{
string line;
for (size_t count = 0; (line = readln) !is null; count++)
{
writeln("Input ", count, ": ", line);
}
}
writeln;
alias (alias) gcd_source_read_pipe.dispatch_queue_t = void*dispatch_queue_t = void*;
alias (alias) gcd_source_read_pipe.dispatch_source_t = void*dispatch_source_t = void*;
alias (alias) gcd_source_read_pipe.dispatch_semaphore_t = void*dispatch_semaphore_t = void*;
alias (alias) gcd_source_read_pipe.dispatch_function_t = extern (C) void function(void*) nothrowdispatch_function_t = extern (C) void function(void*) nothrow;
extern (C) nothrow @nogc
{
// `DISPATCH_SOURCE_TYPE_READ` is `&_dispatch_source_type_read` — an opaque
// descriptor record, referenced only by address.
extern __gshared const ubyte (constant global) const(ubyte) gcd_source_read_pipe._dispatch_source_type_read_dispatch_source_type_read;
(alias) gcd_source_read_pipe.dispatch_queue_t = void*dispatch_queue_t void* gcd_source_read_pipe.dispatch_queue_create(const(char)* label, void* attr) nothrow @nogcdispatch_queue_create(const(char)* (parameter) const(char)* labellabel, void* (parameter) void* attrattr);
(alias) gcd_source_read_pipe.dispatch_source_t = void*dispatch_source_t void* gcd_source_read_pipe.dispatch_source_create(const(void)* type, ulong handle, ulong mask, void* queue) nothrow @nogcdispatch_source_create(const(void)* (parameter) const(void)* typetype, uintptr_t (parameter) ulong handlehandle,
uintptr_t (parameter) ulong maskmask, (alias) gcd_source_read_pipe.dispatch_queue_t = void*dispatch_queue_t (parameter) void* queuequeue);
void void gcd_source_read_pipe.dispatch_source_set_event_handler_f(void* source, extern (C) void function(void*) nothrow handler) nothrow @nogcdispatch_source_set_event_handler_f((alias) gcd_source_read_pipe.dispatch_source_t = void*dispatch_source_t (parameter) void* sourcesource, (alias) gcd_source_read_pipe.dispatch_function_t = extern (C) void function(void*) nothrowdispatch_function_t (parameter) extern (C) void function(void*) nothrow handlerhandler);
void void gcd_source_read_pipe.dispatch_source_set_cancel_handler_f(void* source, extern (C) void function(void*) nothrow handler) nothrow @nogcdispatch_source_set_cancel_handler_f((alias) gcd_source_read_pipe.dispatch_source_t = void*dispatch_source_t (parameter) void* sourcesource, (alias) gcd_source_read_pipe.dispatch_function_t = extern (C) void function(void*) nothrowdispatch_function_t (parameter) extern (C) void function(void*) nothrow handlerhandler);
void void gcd_source_read_pipe.dispatch_set_context(void* object, void* context) nothrow @nogcdispatch_set_context(void* (parameter) void* objectobject, void* (parameter) void* contextcontext);
uintptr_t ulong gcd_source_read_pipe.dispatch_source_get_data(void* source) nothrow @nogcdispatch_source_get_data((alias) gcd_source_read_pipe.dispatch_source_t = void*dispatch_source_t (parameter) void* sourcesource);
uintptr_t ulong gcd_source_read_pipe.dispatch_source_get_handle(void* source) nothrow @nogcdispatch_source_get_handle((alias) gcd_source_read_pipe.dispatch_source_t = void*dispatch_source_t (parameter) void* sourcesource);
void void gcd_source_read_pipe.dispatch_source_cancel(void* source) nothrow @nogcdispatch_source_cancel((alias) gcd_source_read_pipe.dispatch_source_t = void*dispatch_source_t (parameter) void* sourcesource);
void void gcd_source_read_pipe.dispatch_resume(void* object) nothrow @nogcdispatch_resume(void* (parameter) void* objectobject);
void void gcd_source_read_pipe.dispatch_release(void* object) nothrow @nogcdispatch_release(void* (parameter) void* objectobject);
(alias) gcd_source_read_pipe.dispatch_semaphore_t = void*dispatch_semaphore_t void* gcd_source_read_pipe.dispatch_semaphore_create(long value) nothrow @nogcdispatch_semaphore_create(long (parameter) long valuevalue);
long long gcd_source_read_pipe.dispatch_semaphore_wait(void* sema, ulong timeout) nothrow @nogcdispatch_semaphore_wait((alias) gcd_source_read_pipe.dispatch_semaphore_t = void*dispatch_semaphore_t (parameter) void* semasema, ulong (parameter) ulong timeouttimeout);
long long gcd_source_read_pipe.dispatch_semaphore_signal(void* sema) nothrow @nogcdispatch_semaphore_signal((alias) gcd_source_read_pipe.dispatch_semaphore_t = void*dispatch_semaphore_t (parameter) void* semasema);
}
enum (constant) ulong gcd_source_read_pipe.DISPATCH_TIME_FOREVER = 18446744073709551615LUDISPATCH_TIME_FOREVER = ~0UL;
struct (struct) gcd_source_read_pipe.WatchWatch
{
(alias) gcd_source_read_pipe.dispatch_source_t = void*dispatch_source_t (field) void* gcd_source_read_pipe.Watch.sourcesource;
(alias) gcd_source_read_pipe.dispatch_semaphore_t = void*dispatch_semaphore_t (field) void* gcd_source_read_pipe.Watch.wakeupwakeup; // signalled once per event handler invocation
(alias) gcd_source_read_pipe.dispatch_semaphore_t = void*dispatch_semaphore_t (field) void* gcd_source_read_pipe.Watch.cancelledcancelled;
shared int (field) shared(int) gcd_source_read_pipe.Watch.eventsevents;
shared long (field) shared(long) gcd_source_read_pipe.Watch.lastAvailablelastAvailable;
shared long (field) shared(long) gcd_source_read_pipe.Watch.lastReadlastRead;
shared long (field) shared(long) gcd_source_read_pipe.Watch.totalReadtotalRead;
}
__gshared (struct) gcd_source_read_pipe.WatchWatch (__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch;
/// The event handler. Runs on the source's target queue, on a workqueue thread —
/// so it stays allocation-free and reports through `printf` and atomics.
extern (C) void void gcd_source_read_pipe.onReadable(void* context) nothrowThe event handler. Runs on the source's target queue, on a workqueue thread —
so it stays allocation-free and reports through printf and atomics.
onReadable(void* (parameter) void* contextcontext) nothrow
{
import (package) corecore.(package) core.stdcstdc.(module) core.stdc.stdioD header file for C99 <stdio.h>
pubs.opengroup.org/onlinepubs/009695399/basedefs/stdio.h.html, stdio.h
Source
core/stdc/stdio.d
stdio : (alias) printf = int core.stdc.stdio.printf(scope const(char*) format, scope const ...) nothrow @nogcprintf;
// kqueue's `kevent.data` for EVFILT_READ: bytes available right now.
const (local variable) const(long) availableavailable = cast(long) ulong gcd_source_read_pipe.dispatch_source_get_data(void* source) nothrow @nogcdispatch_source_get_data((__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) void* gcd_source_read_pipe.Watch.sourcesource);
const (local variable) const(int) fdfd = cast(int) ulong gcd_source_read_pipe.dispatch_source_get_handle(void* source) nothrow @nogcdispatch_source_get_handle((__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) void* gcd_source_read_pipe.Watch.sourcesource);
char[512] (local variable) char[512] bufferbuffer = void;
const (local variable) const(ulong) wantedwanted = (local variable) const(long) availableavailable > 0 && (local variable) const(long) availableavailable < (local variable) char[512] bufferbuffer.(constant) ulong char[512].length = 512LUlength ? cast((alias) object.size_t = ulongsize_t) (local variable) const(long) availableavailable
: (local variable) char[512] bufferbuffer.(constant) ulong char[512].length = 512LUlength;
const (local variable) const(long) gotgot = (local variable) const(long) availableavailable == 0 ? 0 : long core.sys.posix.unistd.read(int, void*, ulong) nothrow @nogcread((local variable) const(int) fdfd, (local variable) char[512] bufferbuffer.(constant) char* char[512].ptr = &bufferptr, (local variable) const(ulong) wantedwanted);
int core.atomic.atomicOp!("+=", int, int)(ref shared(int) val, int mod) pure nothrow @nogc @safePerforms the binary operation 'op' on val using 'mod' as the modifier.
atomicOp!"+="((__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) shared(int) gcd_source_read_pipe.Watch.eventsevents, 1);
void core.atomic.atomicStore!(MemoryOrder.seq, long, const(long))(ref shared(long) val, const(long) newval) pure nothrow @nogc @trustedWrites 'newval' into 'val'. The memory barrier specified by 'ms' is
applied to the operation, which is fully sequenced by default.
Valid memory orders are MemoryOrder.raw, MemoryOrder.rel, and
MemoryOrder.seq.
atomicStore((__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) shared(long) gcd_source_read_pipe.Watch.lastAvailablelastAvailable, (local variable) const(long) availableavailable);
void core.atomic.atomicStore!(MemoryOrder.seq, long, long)(ref shared(long) val, long newval) pure nothrow @nogc @trustedWrites 'newval' into 'val'. The memory barrier specified by 'ms' is
applied to the operation, which is fully sequenced by default.
Valid memory orders are MemoryOrder.raw, MemoryOrder.rel, and
MemoryOrder.seq.
atomicStore((__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) shared(long) gcd_source_read_pipe.Watch.lastReadlastRead, cast(long) (local variable) const(long) gotgot);
if ((local variable) const(long) gotgot > 0)
long core.atomic.atomicOp!("+=", long, long)(ref shared(long) val, long mod) pure nothrow @nogc @safePerforms the binary operation 'op' on val using 'mod' as the modifier.
atomicOp!"+="((__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) shared(long) gcd_source_read_pipe.Watch.totalReadtotalRead, cast(long) (local variable) const(long) gotgot);
int core.stdc.stdio.printf(scope const(char*) format, scope const ...) nothrow @nogcprintf(" event %d: get_data()=%lld read()=%lld\n",
int core.atomic.atomicLoad!(MemoryOrder.seq, int)(ref return scope shared(const(int)) val) pure nothrow @nogc @trustedLoads 'val' from memory and returns it. The memory barrier specified
by 'ms' is applied to the operation, which is fully sequenced by
default. Valid memory orders are MemoryOrder.raw, MemoryOrder.acq,
and MemoryOrder.seq.
atomicLoad((__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) shared(int) gcd_source_read_pipe.Watch.eventsevents), (local variable) const(long) availableavailable, cast(long) (local variable) const(long) gotgot);
// EOF is not a distinct event: kqueue keeps reporting the descriptor
// readable with zero bytes, and `EV_DISPATCH` re-arms the registration as
// soon as this handler returns. A source that is not cancelled here spins
// at full speed. Cancelling from inside the handler is the only way to stop
// it deterministically.
if ((local variable) const(long) availableavailable == 0)
void gcd_source_read_pipe.dispatch_source_cancel(void* source) nothrow @nogcdispatch_source_cancel((__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) void* gcd_source_read_pipe.Watch.sourcesource);
long gcd_source_read_pipe.dispatch_semaphore_signal(void* sema) nothrow @nogcdispatch_semaphore_signal((__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) void* gcd_source_read_pipe.Watch.wakeupwakeup);
}
/// Runs after the source is fully cancelled — the only point at which the file
/// descriptor may be closed.
extern (C) void void gcd_source_read_pipe.onCancel(void* context) nothrowRuns after the source is fully cancelled — the only point at which the file
descriptor may be closed.
onCancel(void* (parameter) void* contextcontext) nothrow
{
import (package) corecore.(package) core.stdcstdc.(module) core.stdc.stdioD header file for C99 <stdio.h>
pubs.opengroup.org/onlinepubs/009695399/basedefs/stdio.h.html, stdio.h
Source
core/stdc/stdio.d
stdio : (alias) printf = int core.stdc.stdio.printf(scope const(char*) format, scope const ...) nothrow @nogcprintf;
int core.stdc.stdio.printf(scope const(char*) format, scope const ...) nothrow @nogcprintf(" cancel handler: safe to close the descriptor now\n");
long gcd_source_read_pipe.dispatch_semaphore_signal(void* sema) nothrow @nogcdispatch_semaphore_signal((__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) void* gcd_source_read_pipe.Watch.cancelledcancelled);
}
int int D main()main()
{
int[2] (local variable) int[2] fdsfds;
if (int core.sys.posix.unistd.pipe(ref int[2]) nothrow @nogc @trustedpipe((local variable) int[2] fdsfds) != 0)
{
void std.stdio.writeln!string(string __param_0) @safeEquivalent to write(args, '\n'). Calling writeln without
arguments is valid and just prints a newline to the standard
output.
Example
Reads stdin and writes it to stdout with an argument
counter.
import std.stdio;
void main()
{
string line;
for (size_t count = 0; (line = readln) !is null; count++)
{
writeln("Input ", count, ": ", line);
}
}
writeln("SKIP: pipe(2) failed");
return 0;
}
const (local variable) const(int) readEndreadEnd = (local variable) int[2] fdsfds[0], (local variable) const(int) writeEndwriteEnd = (local variable) int[2] fdsfds[1];
auto (local variable) void* queuequeue = void* gcd_source_read_pipe.dispatch_queue_create(const(char)* label, void* attr) nothrow @nogcdispatch_queue_create("dev.sparkles.research.gcd.pipe", null);
scope (exit)
void gcd_source_read_pipe.dispatch_release(void* object) nothrow @nogcdispatch_release((local variable) void* queuequeue);
(__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) void* gcd_source_read_pipe.Watch.wakeupwakeup = void* gcd_source_read_pipe.dispatch_semaphore_create(long value) nothrow @nogcdispatch_semaphore_create(0);
(__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) void* gcd_source_read_pipe.Watch.cancelledcancelled = void* gcd_source_read_pipe.dispatch_semaphore_create(long value) nothrow @nogcdispatch_semaphore_create(0);
(__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) void* gcd_source_read_pipe.Watch.sourcesource = void* gcd_source_read_pipe.dispatch_source_create(const(void)* type, ulong handle, ulong mask, void* queue) nothrow @nogcdispatch_source_create(&(constant global) const(ubyte) gcd_source_read_pipe._dispatch_source_type_read_dispatch_source_type_read, (local variable) const(int) readEndreadEnd, 0, (local variable) void* queuequeue);
void gcd_source_read_pipe.dispatch_set_context(void* object, void* context) nothrow @nogcdispatch_set_context((__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) void* gcd_source_read_pipe.Watch.sourcesource, &(__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch);
void gcd_source_read_pipe.dispatch_source_set_event_handler_f(void* source, extern (C) void function(void*) nothrow handler) nothrow @nogcdispatch_source_set_event_handler_f((__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) void* gcd_source_read_pipe.Watch.sourcesource, &void gcd_source_read_pipe.onReadable(void* context) nothrowThe event handler. Runs on the source's target queue, on a workqueue thread —
so it stays allocation-free and reports through printf and atomics.
onReadable);
void gcd_source_read_pipe.dispatch_source_set_cancel_handler_f(void* source, extern (C) void function(void*) nothrow handler) nothrow @nogcdispatch_source_set_cancel_handler_f((__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) void* gcd_source_read_pipe.Watch.sourcesource, &void gcd_source_read_pipe.onCancel(void* context) nothrowRuns after the source is fully cancelled — the only point at which the file
descriptor may be closed.
onCancel);
// Sources are created suspended; nothing is registered with kqueue until
// the first resume.
void gcd_source_read_pipe.dispatch_resume(void* object) nothrow @nogcdispatch_resume((__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) void* gcd_source_read_pipe.Watch.sourcesource);
void std.stdio.writeln!string(string __param_0) @safeEquivalent to write(args, '\n'). Calling writeln without
arguments is valid and just prints a newline to the standard
output.
Example
Reads stdin and writes it to stdout with an argument
counter.
import std.stdio;
void main()
{
string line;
for (size_t count = 0; (line = readln) !is null; count++)
{
writeln("Input ", count, ": ", line);
}
}
writeln("state 1 — writer writes 11 bytes:");
long core.sys.posix.unistd.write(int, scope const(void*), ulong) nothrow @nogcwrite((local variable) const(int) writeEndwriteEnd, "hello world".(constant) immutable(char)* "hello world".ptr = "hello world"ptr, 11);
long gcd_source_read_pipe.dispatch_semaphore_wait(void* sema, ulong timeout) nothrow @nogcdispatch_semaphore_wait((__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) void* gcd_source_read_pipe.Watch.wakeupwakeup, (constant) ulong gcd_source_read_pipe.DISPATCH_TIME_FOREVER = 18446744073709551615LUDISPATCH_TIME_FOREVER);
assert(long core.atomic.atomicLoad!(MemoryOrder.seq, long)(ref return scope shared(const(long)) val) pure nothrow @nogc @trustedLoads 'val' from memory and returns it. The memory barrier specified
by 'ms' is applied to the operation, which is fully sequenced by
default. Valid memory orders are MemoryOrder.raw, MemoryOrder.acq,
and MemoryOrder.seq.
atomicLoad((__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) shared(long) gcd_source_read_pipe.Watch.lastAvailablelastAvailable) == 11, "EVFILT_READ data was not the byte count");
assert(long core.atomic.atomicLoad!(MemoryOrder.seq, long)(ref return scope shared(const(long)) val) pure nothrow @nogc @trustedLoads 'val' from memory and returns it. The memory barrier specified
by 'ms' is applied to the operation, which is fully sequenced by
default. Valid memory orders are MemoryOrder.raw, MemoryOrder.acq,
and MemoryOrder.seq.
atomicLoad((__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) shared(long) gcd_source_read_pipe.Watch.lastReadlastRead) == 11, "short read");
void std.stdio.writeln!string(string __param_0) @safeEquivalent to write(args, '\n'). Calling writeln without
arguments is valid and just prints a newline to the standard
output.
Example
Reads stdin and writes it to stdout with an argument
counter.
import std.stdio;
void main()
{
string line;
for (size_t count = 0; (line = readln) !is null; count++)
{
writeln("Input ", count, ": ", line);
}
}
writeln("state 2 — writer writes 4 more bytes:");
long core.sys.posix.unistd.write(int, scope const(void*), ulong) nothrow @nogcwrite((local variable) const(int) writeEndwriteEnd, "more".(constant) immutable(char)* "more".ptr = "more"ptr, 4);
long gcd_source_read_pipe.dispatch_semaphore_wait(void* sema, ulong timeout) nothrow @nogcdispatch_semaphore_wait((__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) void* gcd_source_read_pipe.Watch.wakeupwakeup, (constant) ulong gcd_source_read_pipe.DISPATCH_TIME_FOREVER = 18446744073709551615LUDISPATCH_TIME_FOREVER);
assert(long core.atomic.atomicLoad!(MemoryOrder.seq, long)(ref return scope shared(const(long)) val) pure nothrow @nogc @trustedLoads 'val' from memory and returns it. The memory barrier specified
by 'ms' is applied to the operation, which is fully sequenced by
default. Valid memory orders are MemoryOrder.raw, MemoryOrder.acq,
and MemoryOrder.seq.
atomicLoad((__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) shared(long) gcd_source_read_pipe.Watch.lastAvailablelastAvailable) == 4, "second wakeup reported the wrong count");
void std.stdio.writeln!string(string __param_0) @safeEquivalent to write(args, '\n'). Calling writeln without
arguments is valid and just prints a newline to the standard
output.
Example
Reads stdin and writes it to stdout with an argument
counter.
import std.stdio;
void main()
{
string line;
for (size_t count = 0; (line = readln) !is null; count++)
{
writeln("Input ", count, ": ", line);
}
}
writeln("state 3 — writer closes its end (the handler cancels the source):");
int core.sys.posix.unistd.close(int) nothrow @nogc @trustedclose((local variable) const(int) writeEndwriteEnd);
long gcd_source_read_pipe.dispatch_semaphore_wait(void* sema, ulong timeout) nothrow @nogcdispatch_semaphore_wait((__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) void* gcd_source_read_pipe.Watch.wakeupwakeup, (constant) ulong gcd_source_read_pipe.DISPATCH_TIME_FOREVER = 18446744073709551615LUDISPATCH_TIME_FOREVER);
assert(long core.atomic.atomicLoad!(MemoryOrder.seq, long)(ref return scope shared(const(long)) val) pure nothrow @nogc @trustedLoads 'val' from memory and returns it. The memory barrier specified
by 'ms' is applied to the operation, which is fully sequenced by
default. Valid memory orders are MemoryOrder.raw, MemoryOrder.acq,
and MemoryOrder.seq.
atomicLoad((__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) shared(long) gcd_source_read_pipe.Watch.lastAvailablelastAvailable) == 0, "EOF wakeup carried a non-zero byte count");
assert(long core.atomic.atomicLoad!(MemoryOrder.seq, long)(ref return scope shared(const(long)) val) pure nothrow @nogc @trustedLoads 'val' from memory and returns it. The memory barrier specified
by 'ms' is applied to the operation, which is fully sequenced by
default. Valid memory orders are MemoryOrder.raw, MemoryOrder.acq,
and MemoryOrder.seq.
atomicLoad((__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) shared(long) gcd_source_read_pipe.Watch.lastReadlastRead) == 0, "read at EOF returned data");
// Cancellation is asynchronous even when requested from the handler: the
// cancel handler is the completion signal, and the only point at which the
// descriptor may be closed.
long gcd_source_read_pipe.dispatch_semaphore_wait(void* sema, ulong timeout) nothrow @nogcdispatch_semaphore_wait((__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) void* gcd_source_read_pipe.Watch.cancelledcancelled, (constant) ulong gcd_source_read_pipe.DISPATCH_TIME_FOREVER = 18446744073709551615LUDISPATCH_TIME_FOREVER);
int core.sys.posix.unistd.close(int) nothrow @nogc @trustedclose((local variable) const(int) readEndreadEnd);
void gcd_source_read_pipe.dispatch_release(void* object) nothrow @nogcdispatch_release((__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) void* gcd_source_read_pipe.Watch.sourcesource);
void std.stdio.writeln!()() @safeEquivalent to write(args, '\n'). Calling writeln without
arguments is valid and just prints a newline to the standard
output.
Example
Reads stdin and writes it to stdout with an argument
counter.
import std.stdio;
void main()
{
string line;
for (size_t count = 0; (line = readln) !is null; count++)
{
writeln("Input ", count, ": ", line);
}
}
writeln();
void std.stdio.writefln!(char, int, long)(in char[] fmt, int __param_1, long __param_2) @safeEquivalent to writef(fmt, args, '\n').
writefln("handler invocations: %d, bytes delivered: %d",
int core.atomic.atomicLoad!(MemoryOrder.seq, int)(ref return scope shared(const(int)) val) pure nothrow @nogc @trustedLoads 'val' from memory and returns it. The memory barrier specified
by 'ms' is applied to the operation, which is fully sequenced by
default. Valid memory orders are MemoryOrder.raw, MemoryOrder.acq,
and MemoryOrder.seq.
atomicLoad((__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) shared(int) gcd_source_read_pipe.Watch.eventsevents), long core.atomic.atomicLoad!(MemoryOrder.seq, long)(ref return scope shared(const(long)) val) pure nothrow @nogc @trustedLoads 'val' from memory and returns it. The memory barrier specified
by 'ms' is applied to the operation, which is fully sequenced by
default. Valid memory orders are MemoryOrder.raw, MemoryOrder.acq,
and MemoryOrder.seq.
atomicLoad((__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) shared(long) gcd_source_read_pipe.Watch.totalReadtotalRead));
assert(int core.atomic.atomicLoad!(MemoryOrder.seq, int)(ref return scope shared(const(int)) val) pure nothrow @nogc @trustedLoads 'val' from memory and returns it. The memory barrier specified
by 'ms' is applied to the operation, which is fully sequenced by
default. Valid memory orders are MemoryOrder.raw, MemoryOrder.acq,
and MemoryOrder.seq.
atomicLoad((__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) shared(int) gcd_source_read_pipe.Watch.eventsevents) == 3, "expected exactly three wakeups");
assert(long core.atomic.atomicLoad!(MemoryOrder.seq, long)(ref return scope shared(const(long)) val) pure nothrow @nogc @trustedLoads 'val' from memory and returns it. The memory barrier specified
by 'ms' is applied to the operation, which is fully sequenced by
default. Valid memory orders are MemoryOrder.raw, MemoryOrder.acq,
and MemoryOrder.seq.
atomicLoad((__gshared global) gcd_source_read_pipe.Watch gcd_source_read_pipe.watchwatch.(field) shared(long) gcd_source_read_pipe.Watch.totalReadtotalRead) == 15, "expected 15 bytes across the two data wakeups");
return 0;
}