The pipe is the apex interprocess communication mechanism. With a pipe, data can flow from one program to the next without any extra code in those programs. It is likely the reason why an experimental operating system from 1969 (UNIX), has outlived all others.
Fundamentally a pipe is nothing more then a decorated ring buffer. The ring buffer is the datastructure which holds the data flowing trough the pipe. The decorations are things like its locks, page mappings, and file descriptors.
/dev/pipe
I am not going to make pipelines via the shell notation ($ foo | bar | baz); these pipes are unnamed/anonymous.
To make the examples more clear I am going to create a named pipe, and because a pipe behaves like a character device I am going to put it in /dev.
# mkfifo -m a=rw /dev/pipe
Figure: Creating a named pipe.
First I am going to compare our named pipe with an anonymous pipe, by executing the same shell sequence.
The output won’t be exactly the same, because we are reading from /dev/random.
The example using the named pipe it split into two parts, a write part and a read part.
$ dd if=/dev/random of=/dev/stdout bs=8 count=2 | od -h
2+0 records in
2+0 records out
16 bytes transferred in 0.001 secs (16000 bytes/sec)
0000000 b844 04c3 a180 16d6 0389 67f4 ee82 1f4f
0000020
Figure: Sequence with anonymous pipe.
$ dd if=/dev/random of=/dev/pipe bs=8 count=2
2+0 records in
2+0 records out
16 bytes transferred in 0.001 secs (16000 bytes/sec)
Figure: Write sequence with named pipe.
$ od -h /dev/pipe
0000000 4aca 4894 7955 6bf2 8a5c d757 3237 a220
0000020
Figure: Read sequence with named pipe.
During the execution of these sequences, the reader (od) is initially being blocked because there is nothing streaming trough the pipe.
But when the writer (dd) is writing to the pipe the pipe unblocks and the data flows, allowing the reader to read.
Reading from a blocking pipe is be fun, but what if we don’t want to be blocked?
fstrp(1)
In a previous post I wrote that I wanted to practice event programming. During the writeup of this post it occurred to me that pipes are the a great way to learn event programming; especially because Kqueue (the BSD event notification interface) is such a renowned part of the system.
In short: fstrp tries to find a string while reading a pipe. If it does find the specified string, it will exit and it won’t write anything. If it does not find the specified string, it will write the content of the pipe.
package main
/*
Fstrp: Find string in pipe
0 => Everything went well and the string was found.
1 => Something went wrong (like I/O operation), prints error to stderr.
2 => The string was not found, print content to stderr.
*/
import (
"fmt"
"golang.org/x/sys/unix"
"io"
"os"
"strings"
)
func main() {
var (
str string
fp int // File descriptor
fi os.FileInfo // File info
fa *os.File // File abstraction, wrapper over fp
err error
kq int // Kqueue descriptor
kev unix.Kevent_t // Kqueue event
evl = make([]unix.Kevent_t, 1)
buf []byte // Buffer
)
// Make sure all the arguments are met
if len(os.Args) != 3 {
fmt.Fprintln(os.Stderr, "usage: fstrp [string] [pipe]")
os.Exit(1)
}
str = os.Args[1]
// Make sure that the specified file is an actual pipe
if fi, err = os.Stat(os.Args[2]); err != nil {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
} else if fi.Mode()&os.ModeNamedPipe == 0 {
fmt.Fprintln(os.Stderr, "File is not a pipe")
os.Exit(1)
}
// Open the pipe in non-blocking mode
// Wrap the *os.File around the descriptor
if fp, err = unix.Open(os.Args[2], unix.O_RDONLY|unix.O_NONBLOCK, 0); err != nil {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
fa = os.NewFile(uintptr(fp), "rpipe")
defer fa.Close()
// Allocate a Kqueue
if kq, err = unix.Kqueue(); err != nil {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
defer unix.Close(kq)
// Define an read event on the pipe for kqueue
kev = unix.Kevent_t{
Ident: uint64(fp),
Filter: unix.EVFILT_READ,
Flags: unix.EV_ADD | unix.EV_ENABLE | unix.EV_CLEAR,
Fflags: 0,
Data: 0,
Udata: 0,
}
for {
var n int
// Wait until the pipe is ready for reading
if n, err = unix.Kevent(kq, []unix.Kevent_t{kev}, evl, nil); err != nil {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
if n <= 0 {
fmt.Fprintln(os.Stderr, "Invalid event triggered")
continue
}
// Read content of the pipe into the buffer
if buf, err = io.ReadAll(fa); err != nil {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
break
}
// Find specified string in buffer
if strings.Contains(string(buf), str) {
os.Exit(0)
}
fmt.Fprint(os.Stdout, string(buf))
os.Exit(2)
}
Figure: The source code of fstrp(1).
I am not going to explain the program here, as the source code commentary provides good enough explanation. In short: Instead of waiting for pipe I/O directly, we utilize Kqueue to get notified when the pipe is ready for I/O.
The ultimate design tool
Pipes are amazing. I think that to write programs for UNIX-like systems, your program should be able to handle reading from and writing to pipes, even more so than files. There are also many more things that can break while handling pipes, SIGPIPE for example. I believe the inverse is also true: If your program breaks because it can’t handle pipes, your program is not a true “UNIX” program. So handle your pipes, and let your program be great.
References
- Wikipedia Unix link
- NetBSD pipe(2) link
- Wikipedia Circular buffer link
- GitHub/NetBSD src/sys/sys/pipe.h link
- GitHub/NetBSD src/sys/kern/sys_pipe.c link
- NetBSD fd(4) link
- NetBSD mkfifo(1) link
- NetBSD random(4) link
- NetBSD dd(1) link
- NetBSD od(1) link
- NetBSD kqueue(2) link
- Wikipedia Event-driven programming link
- GitHub/TheDevtop fstrp(1) link
- NetBSD signal(7) link