Communicating Sequential Processes (CSP)

Background and Context

Tony Hoare made the following observations about concurrent systems:

• Having multiple processes share the same state is problematic because it is
  • Prone to race conditions, requiring careful synchronization, which is
  • Challenging to code correctly or reason about, and
  • Potentially inefficient due to contention.
• Writing code with explicit lock-based synchronization is
  • Unnatural, and
  • Error-prone.

He came up with a new vision, Communicating Sequential Processes (CSP), to address these issues. The key principles of CSP include

1. Processes are independent and do not share state.
2. Communication between processes occurs via messages sent over channels.
3. Synchronization is achieved through message passing, not locks.

In fact, when Hoare spoke of “processes” he meant independent threads of execution, not necessarily separate programs. These processes can be lightweight and managed by a single program, as in Go's “goroutines”.

The key thing is that CSP represented a significant shift in thinking about concurrency, moving away from shared memory and lock-based synchronization towards a model of independent processes that communicate through channels.

CSP in Modern Programming

While pure CSP implementations are relatively rare, its ideas have been hugely influential. Some notable examples include

• Erlang's Actor model (1986); also adopted by Scala and Swift.
• Occam-π, a modern variant of occam (2005)
• Clojure's core.async library (2013)
• Go's concurrency model (2009)

Go, in particular, has brought CSP-style concurrency to mainstream programming. While not a pure implementation of CSP, Go's goroutines and channels provide a practical and efficient way to apply CSP principles in a general-purpose programming language.

Producer–Consumer Problem in Go

Let's implement the producer–consumer problem using Go's concurrency primitives. Skim the code below and then give it a try in Online GDB.

package main

import (
    "fmt"
    "math/rand"
    "sync"
    "time"
)

const (
    bufferSize          = 32
    numProducers        = 2
    numConsumers        = 2
    messagesPerProducer = 10
)

type message struct {
    content string
}

func producer(id int, messages chan<- message, wg *sync.WaitGroup) {
    defer wg.Done()
    for i := 0; i < messagesPerProducer; i++ {
        msg := message{fmt.Sprintf("Message %d from Producer %d", i, id)}
        messages <- msg
        time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
    }
    messages <- message{"Bye"}
}

func consumer(id int, messages <-chan message, wg *sync.WaitGroup) {
    defer wg.Done()
    for msg := range messages {
        fmt.Printf("Consumer %d: %s\n", id, msg.content)
        if msg.content == "Bye" {
            return
        }
        time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
    }
}

func main() {
    messages := make(chan message, bufferSize)
    var wg sync.WaitGroup

    // Start producers
    for i := 0; i < numProducers; i++ {
        wg.Add(1)
        go producer(i, messages, &wg)
    }

    // Start consumers
    for i := 0; i < numConsumers; i++ {
        wg.Add(1)
        go consumer(i, messages, &wg)
    }

    // Wait for all goroutines to finish
    wg.Wait()
    close(messages)
}

• Try this code in Online GDB

You can contrast this code against the C implementation we saw in Week 3, Lesson 1.

• Check out the old Locks and CVs C code in Online GDB

Understanding the Go Code

Let's look at a few key lines from the code.

messages := make(chan message, bufferSize)

This line creates a channel named messages that can hold up to bufferSize messages. Channels are typed, thread-safe queues that allow goroutines to communicate. This channel is known as a buffered channel as it receive messages (up to a limit) even when no one is ready to read them.

Go also allows unbuffered channels, which block until both a sender and receiver are ready to communicate. We would make such a channel by saying messages := make(chan message).

go producer(i, messages, &wg)

The go keyword launches the producer function as a goroutine. Goroutines are lightweight threads managed by the Go runtime.

msg := message{fmt.Sprintf("Message %d from Producer %d", i, id)}
messages <- msg

This code creates a new message struct and sends it to the messages channel. The <- operator is used to send and receive messages on a channel.

for msg := range messages {
    // Process messages
}

The range keyword is used to iterate over messages received from the messages channel. The loop continues until the channel is closed. We could also receive an individual message using msg := <-messages.

close(messages)

This line closes the messages channel, signaling that no more messages will be sent. It's essential to close channels to avoid goroutines blocking indefinitely.

The sync.WaitGroup type is used to wait for a collection of goroutines to finish. It acts as a counter tracking the number of tasks in progress. The Add method increments the counter, Done decrements it, and Wait blocks until the counter reaches zero. Thus, in the code, we see

wg.Add(numProducers)

This line increments the WaitGroup counter by the number of producers. Each producer will call wg.Done() when it finishes, decrementing the counter.

wg.Wait()

This line blocks until all producers have finished, ensuring that the program doesn't exit prematurely.

This concept is sometimes called a barrier.

Implementation

Although we can implement communications channels in shared memory using traditional locks and condition variables, they aren't the only option. CSP scales to distributed systems, where the communication is literally over a network. The principles of CSP can be applied at many levels of abstraction.

The Power of select

In this version we've added one new goroutine, a printer that handles all printing. It has three input channels: one for messages from the producers, one for messages from the consumers, and one for a quit signal. It uses a select statement to choose between these channels, but also complains if it hasn't gotten a message in a while.

Let's take a look at the code:

package main

import (
    "fmt"
    "math/rand"
    "sync"
    "time"
)

const (
    bufferSize          = 32
    numProducers        = 2
    numConsumers        = 2
    messagesPerProducer = 10
)

type message struct {
    content string
}

type printMessage struct {
    source  string
    content string
}

func producer(id int, messages chan<- message, printer chan<- printMessage, wg *sync.WaitGroup) {
    defer wg.Done()
    for i := 0; i < messagesPerProducer; i++ {
        msg := message{fmt.Sprintf("Message %d from Producer %d", i, id)}
        messages <- msg
        printer <- printMessage{"producer", fmt.Sprintf("Producer %d sent: %s", id, msg.content)}
        time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
    }
    messages <- message{"Bye"}
}

func consumer(id int, messages <-chan message, printer chan<- printMessage, wg *sync.WaitGroup) {
    defer wg.Done()
    for msg := range messages {
        printer <- printMessage{"consumer", fmt.Sprintf("Consumer %d received: %s", id, msg.content)}
        if msg.content == "Bye" {
            return
        }
        time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
    }
}

func printer(producerMsgs <-chan printMessage, consumerMsgs <-chan printMessage, done <-chan bool) {
    const TIMEOUT_MS = 2000

    for {
        select {
        case msg := <-producerMsgs:
            fmt.Printf("[PRODUCER] %s\n", msg.content)
        case msg := <-consumerMsgs:
            fmt.Printf("[CONSUMER] %s\n", msg.content)
        case <-done:
            fmt.Println("Printer: Shutting down...")
            return
        case <-time.After(TIMEOUT_MS * time.Millisecond):
            fmt.Println("Printer: No activity for", TIMEOUT_MS, "ms. Feeling lonely...")
        }
    }
}

func main() {
    rand.Seed(time.Now().UnixNano())
    messages := make(chan message, bufferSize)
    producerMsgs := make(chan printMessage, bufferSize)
    consumerMsgs := make(chan printMessage, bufferSize)
    done := make(chan bool)

    var wg sync.WaitGroup

    // Start printer
    go printer(producerMsgs, consumerMsgs, done)

    // Start producers
    for i := 0; i < numProducers; i++ {
        wg.Add(1)
        go producer(i, messages, producerMsgs, &wg)
    }

    // Start consumers
    for i := 0; i < numConsumers; i++ {
        wg.Add(1)
        go consumer(i, messages, consumerMsgs, &wg)
    }

    // Wait for all goroutines to finish
    wg.Wait()
    close(messages)

    // Signal printer to shut down
    done <- true

    // Allow some time for final messages to be printed
    time.Sleep(100 * time.Millisecond)
}

• Try this code in Online GDB

Key Points

1. Channels: The messages channel replaces the circular buffer. It's a typed, thread-safe queue provided by Go.
2. Goroutines: producer and consumer functions run as goroutines, lightweight threads managed by the Go runtime.
3. No Explicit Locking: There are no mutexes or condition variables. Synchronization is handled implicitly by channel operations.
4. Simplicity: The Go code is significantly simpler than the C version, with no need for explicit buffer management or lock handling.

Comparing CSP (Go) and Lock-based (C) Approaches

1. Abstraction: Go's channels provide a higher level of abstraction, hiding the complexities of synchronization.
2. Safety: The CSP approach reduces the risk of deadlocks and race conditions by design.
3. Scalability: Go's runtime can efficiently manage many goroutines, making it easier to scale to many producers and consumers.
4. Readability: The Go code is generally more straightforward and easier to reason about.
5. Flexibility: While this example is similar in structure to the C version, CSP allows for more flexible communication patterns that would be complicated to implement with locks.

Conclusion

CSP and Go's implementation of its principles offer a powerful alternative to traditional lock-based concurrency. By focusing on communication rather than shared state, CSP can lead to more robust, scalable, and understandable concurrent programs. However, it's important to note that both approaches have their place, and understanding both is valuable for any programmer working with concurrent systems.

(When logged in, completion status appears here.)