Building a Distributed System in Go: Part 1 — In-Process Message Passing & CSP Primitives

Welcome to Part 1 of the Go Distributed Systems Lab series! Over the course of 20 hands-on projects, we are building core distributed systems primitives from the ground up using Go 1.22+ and the standard library (net, sync, context, log/slog, encoding/binary).

Before jumping into raw socket framing, gossip protocols, or Raft consensus, we need to master the foundational concurrency building blocks inside a single process: Goroutines, Channels, and Communicating Sequential Processes (CSP).

💡 The Philosophy: Share Memory by Communicating

In traditional concurrent programming (like C++ or Java), thread synchronization often relies on shared memory protected by mutexes, lock-free queues, or read-write locks.

Go flips this model with a core design principle:

"Do not communicate by sharing memory; instead, share memory by communicating."

By passing ownership of data structures through Go channels, each pipeline stage operates on isolated memory. This eliminates data races by design without requiring explicit lock management (sync.Mutex).

🏗️ Architecture & Component Design

In this first module (01-message-passing), we construct a 3-stage data processing pipeline:

+------------------+         Job Channel          +------------------+        Result Channel       +-------------------+
|     Producer     |  ------------------------->  |      Worker      |  ------------------------>  |     Collector     |
| (Generates Jobs) |     (Buffered, cap=10)       | (Isolated State) |    (Buffered, cap=10)      | (Aggregates Data) |
+------------------+                              +------------------+                             +-------------------+

1. Ingestion Stage (Producer)

Generates typed Job values and pushes them into a direction-constrained buffered channel (chan<- Job). When generation finishes, it closes the channel to broadcast an end-of-stream signal.

2. Processing Stage (Worker)

Consumes from <-chan Job using Go's for job := range in construct. The worker maintains internal execution metrics (e.g., processedCount) entirely within its local stack scope—no locks required.

3. Collector Stage

Receives output items from <-chan Result and processes them asynchronously until the channel is closed.

💻 Full Implementation

Here is the complete standard-library implementation using Go 1.22+:

package main

import (
    "fmt"
    "log/slog"
    "os"
    "sync"
    "time"
)

// Job represents a unit of work passed between isolated pipeline stages.
type Job struct {
    ID        int
    Payload   string
    Processed bool
}

// Result represents the output produced by the processor stage.
type Result struct {
    JobID     int
    Output    string
    Timestamp time.Time
}

func main() {
    // Initialize standard structured logger (log/slog)
    logger := slog.New(slog.NewTextHandler(os.Stdout, nil))
    slog.SetDefault(logger)

    slog.Info("Starting message passing pipeline", "stage", "initialization")

    // Channels acting as bounded queues
    jobs := make(chan Job, 10)
    results := make(chan Result, 10)

    var wg sync.WaitGroup

    // Stage 1: Ingestion Producer
    wg.Add(1)
    go producer(jobs, 5, &wg)

    // Stage 2: Processing Worker (Isolated State Pipeline)
    wg.Add(1)
    go worker(1, jobs, results, &wg)

    // Stage 3: Aggregator / Collector
    done := make(chan struct{})
    go collector(results, done)

    // Synchronize pipeline termination
    wg.Wait()
    close(results) // Signal collector to exit once upstream workers complete

    <-done // Block until collector drains remaining results
    slog.Info("Pipeline processing complete", "status", "success")
}

// producer generates work items and sends them through the buffered channel.
func producer(out chan<- Job, count int, wg *sync.WaitGroup) {
    defer wg.Done()
    defer close(out) // Closing channel signals EOF to downstream consumers

    for i := 1; i <= count; i++ {
        job := Job{
            ID:      i,
            Payload: fmt.Sprintf("task-payload-%d", i),
        }
        slog.Info("Producing job", "job_id", job.ID)
        out <- job
        time.Sleep(50 * time.Millisecond)
    }
}

// worker encapsulates isolated processing state without shared memory locks.
func worker(id int, in <-chan Job, out chan<- Result, wg *sync.WaitGroup) {
    defer wg.Done()

    // Isolated state local to this goroutine - thread-safe without mutexes!
    processedCount := 0

    for job := range in {
        processedCount++
        slog.Info("Worker processing job", 
            "worker_id", id, 
            "job_id", job.ID, 
            "local_total", processedCount,
        )

        // Simulate computation / processing overhead
        time.Sleep(100 * time.Millisecond)

        out <- Result{
            JobID:     job.ID,
            Output:    fmt.Sprintf("PROCESSED[%s]", job.Payload),
            Timestamp: time.Now(),
        }
    }
}

// collector receives final stage outputs until the stream is exhausted.
func collector(in <-chan Result, done chan<- struct{}) {
    for res := range in {
        slog.Info("Result collected", 
            "job_id", res.JobID, 
            "output", res.Output, 
            "time", res.Timestamp.Format(time.RFC3339Nano),
        )
    }
    close(done)
}

🔍 Key Go Concurrency Patterns Explained

1. Directional Channel Types

Notice the parameter signatures in our pipeline functions:

• out chan<- Job: Send-only channel. Writing into this is allowed; reading or closing from inside the caller scope (if not intended) produces a compile-time error.
• in <-chan Job: Receive-only channel. The worker can only consume from it.

Restricting channel direction at API boundaries prevents accidental closed-channel writes or unauthorized channel closures.

2. Clean Channel Drain Mechanics

When producer() executes close(out), it doesn't delete existing values inside the buffer. Instead, it marks the channel as closed.

Downstream in worker():

for job := range in { ... }

The for range loop continuously extracts items until the channel is empty AND closed, at which point the loop cleanly exits.

3. Structured Logging with log/slog

Since Go 1.21, log/slog provides structured key-value logging natively. In distributed systems, plain text strings become impossible to query. Using slog.Info("...", "job_id", job.ID) ensures consistent log parsing across asynchronous routines.

⚠️ Architectural Limitations of In-Process Message Passing

While in-process channels provide clean abstractions, they have clear boundaries when designing real-world distributed systems:

1. Process Boundary Restrictions: Channels are strictly in-memory data structures managed by the Go runtime scheduler (m:n goroutine multiplexing). They cannot cross host, network, or process boundaries.
2. Volatile Memory: If the application panics or crashes, all messages currently sitting inside buffered channels are lost permanently.
3. Coordinated Backpressure: Channels handle backpressure synchronously (blocking on send when full). However, if an upstream system floods the channel faster than workers can consume, memory usage grows up to channel capacity before blocking cascades backward.
4. No Error Return Channels: This basic pattern lacks a mechanism for workers to report execution errors or request task retries.

🚀 What's Next?

In Part 2, we will address the asynchronous communication problem by building a Correlated Request/Reply mechanism. We will implement unique correlation IDs, response channels, and request multiplexing to turn one-way pipelines into interactive distributed calls!

What concurrency pattern do you use most in your Go services? Drop a comment below! 👇

Source code of the same can be found at : https://github.com/pckrishnadas88/go-distributed-systems-lab/tree/main/01-message-passing