Table of Contents

• What are pipes in IPC?
• Why are pipes important?
• Pipes vs Go channels
• When to use pipes
• When to use channels
• Pipes in go
• Anonymous pipes
• Named pipes (FIFOs)
• Implementing

This article was made for a youtube video, if you want to check that out: https://www.youtube.com/watch?v=Ue2l3ZWOAT0

Nowadays, we have all kinds of complex tools in distributed systems, like Kafka or RabbitMQ, message brokers everywhere that help systems talk to each other.

But have you ever wondered: what simple idea made all of this possible in the first place?

Before distributed systems and cloud infrastructure… It all started with a basic communication pattern: a producer sending data to a consumer.

And at the OS level, that pattern is powered by something called Inter-Process Communication, or IPC.

In this video, we’re going all the way back to the basics — and I’ll show you how pipes work to connect two processes together and let them share data, after that we will write some code breaking it down step by step.

Let ‘s jump in.

What are pipes in IPC?

Although we can use other tools to perform IPCs, such as sockets or shared memory, we will focus more on pipes on this video.

We can think of pipes as a message tube that connects different programs or processes running inside a computer.

Linux users use pipes all the time when using terminal, such as:

cat file.txt | grep "hello world"

Just like how you might put a message in one end of a tube and someone receives it at the other end, pipes let one program (we call this program “producer”) send information through this virtual tube to another program (the “consumer”) that is waiting to receive it.

Think about it like that: Program A has some data it wants to share, it drops the data into the pipe and Program B picks up that data from the other end of the pipe. It’s a simple but powerful way for programs to talk to each other and work together while they’re running separately on your computer.

This workflow follows the same principles of the producer-consumer model.

A key element of IPC, pipes establish a unidirectional flow of information, which basically means that data consistently moves in only one direction from the “write end”to the “read end” of the pipe.

Pipes are used in a variety of system-level programming tasks. Such as:

• Data streaming: When data needs to be streamed from one process to another, pipes offer a simple and effective solution.

• Inter process data exchange: pipes facilitate data exchange between processes, essential in many multi process applications.

• Command-line utilities: pipes are often used to connect the output of one command-line utility to the input of another, this way we can create powerful command chains.

Why are pipes important?

Modularity: They allow specialized programs that can build something well and chain the data to another efficiently. Instead of one massive program, you get reusable components.

Efficiency: Data flows directly between processes - no temporary files or intermediate storage are used. The OS handles the plumbing.

Speed mismatches: Built-in buffering lets a fast writer keep going even if the reader is slow (or vice versa). No blocking, no data loss.

Simple but powerful: One line of code in your terminal, connects complex operations.

Pipes vs Go channels

You might be thinking that pipes are basically like Go channels, and you’re not wrong - they do share some similarities:

What they have in common:

•  They’re both used for communication: Pipes let different programs talk to each other, while channels let goroutines communicate within your Go program.

 - Data transfer: Whether it’s between processes (pipes) or goroutines (channels), both are about getting information from point A to point B.

• Synchronization: Both provide a level of synchronization which means that they can make things wait. Writing to a full pipe or reading from an empty pipe will block the process until the pipe is empty or and data is sent to it. In a similar way, receiving from an empty channel in Go will block the goroutine until the channel is ready for more data.

• Buffering: Pipes and channels can be buffered, which means they can store data before the receiver is ready. A buffered pipe has a defined capacity before it blocks or overflows, similarly, Go channels can be created with a capacity, allowing a certain amount of data to be held without immediate receiver readiness.

Differences:

• Direction of communication: regular pipes only go one way (like a one-way street), but Go channels work both ways by default. Channels in Go are bidirectional by default, allowing data to be sent and received on the same channel.

• Ease of use in context: channels are built into Go and just work, while pipes need more setup since they’re an operating system feature. ****

When to use pipes

• You need different programs or processes to talk to each other (maybe even programs written in different languages). 

• You’re working with separate applications that need to share data.

• You’re on Linux/Unix and want to use the system’s built-in communication tools.

When to use channels

• You’re writing concurrent Go code and need goroutines to coordinate communication.

• You want a simple, safe way to handle concurrency in your Go program.

• You’re building complex concurrency patterns like worker pools or fan-in or fan-out operations.

Pipes in go

Go makes working with pipes pretty straightforward through its standard library. In Go, you will face 2 different type of pipes.

In-memory pipes with io.Pipe(): 

- The io.Pipe() function is commonly used to create a synchronous, in-memory pipe, which means that it lives entirely in your program’s memory.

• It’s synchronous, meaning when you write data, it goes directly to the reader without involving the operating system. It’s just a fancy way to connect two channels (read and write) underneath your code.

Check the source code: https://cs.opensource.google/go/go/+/master:src/io/pipe.go

OS-level pipes with os.Pipe(): 

• When you need “real” pipes that the operating system manages, use os.Pipe(), which internally uses the SYS_PIPE2 syscall and the standard library handles all the complexity underneath for us.

Check the source code: https://cs.opensource.google/go/go/+/master:src/os/pipe_unix.go

How they work: Both types work the same way from your code’s perspective - you write data to one end using normal write operations and read from the other end using standard read operations. Think of it like putting a letter in one end of a tube and someone else pulling it out the other end.

Don’t forget error handling: Just like with any I/O operation, things can go wrong - pipes can break, data can get corrupted, or processes might disappear unexpectedly. Always check for errors to keep your program robust.

If you think you are learning something out of this video, subscribe and leave a like bellow and I promise on the next video you will learn more.

Anonymous pipes

Anonymous pipes are the simplest type of pipes - they are temporary connections that only exist while your program is running. They are used for communication between parent and child processes and once the process ends, the pipe disappears.

A simple example of anonymous pipes for unix users might be a system usage of commands on terminal:

cat file.txt | grep "hello world"

Let’s see a simple example on how to make anonymous pipes.

    func main() {
       reader, writer, _ := os.Pipe()
       cmd := exec.Command("cat")
       // Set the command's standard input to the pipe
       cmd.Stdin = reader
       cmd.Stdout = os.Stdout
       cmd.Start()
       writer.WriteString("Hello from child!\n")
       writer.Close()
       cmd.Wait()
    }

Named pipes (FIFOs)

Unlike anonymous pipes, named pipes are not limited to live processes, they stay in your filesystem even after programs finish. Any process can connect to them by name, making them useful for communication between completely unrelated programs. 

IPC can sometimes be an abstract concept, challenging to understand for those new to system programming. Let’s use a simple, relatable analogy to make this easier.

Picture a busy Coffee shop order system. Customers (producers) come up to place their orders, which get written and lined up in a queue. The baristas (consumers of this order system) grab the order from the queue (pipe) one by one to make the drinks.

During morning rush, the orders pile up faster than the baristas can make the drinks - that’s the full buffer, and the customers might have to wait for this buffer to free up some space ****to place more orders. During quiet afternoon hours, baristas stand around waiting for new orders to appear - that’s an empty buffer.

The order queue (pipe) keeps everything organized and handles the timing mismatches between when people want to order and how fast baristas can work. Whether it’s crazy busy or dead quiet, the system keeps flowing smoothly.

Implementing

Following, it’s an example of how a coffee shop could use this consumer producer model using IPC.

You can check the video here: https://www.youtube.com/watch?v=Ue2l3ZWOAT0&t=756s

Code: https://github.com/viquitorreis/pipes_in_go

func main() {
	coffeeShopPipePath := "/tmp/coffee_shop_pipe"
	if !namedPipeExists(coffeeShopPipePath) {
		if err := unix.Mkfifo(coffeeShopPipePath, 0666); err != nil {
			fmt.Printf("errror while creating pipe: %v\n", err)
			return
		}
	}

	coffeeOrders, err := os.OpenFile(coffeeShopPipePath, os.O_RDWR, os.ModeNamedPipe)
	if err != nil {
		fmt.Printf("error while opening named pipe: %+v", err)
		return
	}
	defer coffeeOrders.Close()

	timeout := time.After(10 * time.Second)

	wg := &sync.WaitGroup{}
	wg.Add(2)
	go produceOrders(wg, coffeeOrders, []string{"Espresso", "Arabica", "Latte", "Cappuccino"})
	go consumeOrders(wg, coffeeOrders, timeout)
	wg.Wait()
}

func namedPipeExists(pipePath string) bool {
	info, err := os.Stat(pipePath)
	if err != nil {
		return false
	}
	return info.Mode()&os.ModeNamedPipe != 0
}

func produceOrders(wg *sync.WaitGroup, coffeOrders *os.File, orders []string) {
	defer wg.Done()
	for _, order := range orders {
		_, err := coffeOrders.WriteString(order + "\n")
		if err != nil {
			fmt.Printf("err while writing to named pipe: %+v", err)
		}
	}
	coffeOrders.WriteString("finish\n")
}

func consumeOrders(wg *sync.WaitGroup, coffeeOrders *os.File, timeout <-chan time.Time) {
	defer wg.Done()
	scanner := bufio.NewScanner(coffeeOrders)
	for scanner.Scan() {
		order := scanner.Text()
		if order == "finish" {
			println("finishing coffee orders.")
			break
		}

		select {
		case <-timeout:
			fmt.Println("Timeout reached, stopping processing orders.")
			return
		default:
			fmt.Println("Processing order: ", order)
			time.Sleep(time.Second * 4)
		}
	}

	if err := scanner.Err(); err != nil {
		fmt.Println("error reading from named pipe: ", err.Error())
	}
}