Session 15¶

Topic 2: The Go language¶

Today¶

Case studies with concurrency¶

caching function calls
chat server

Common pitfalls / concurrency bugs¶

watch out for:

race conditions
deadlocks
goroutine leaks
blocking channels

Case study: caching function calls¶

Problem: Caching results of repeated function calls¶

common problem in real-world concurrent programs:

caching the result of a function so that it need be computed only once
but accessing the cache with many goroutines will create race conditions

Goal: concurrent nonblocking cache¶

cache avoids repeated unnecessary function calls
the cache is accessed by many readers
occasional writes (updates/insertions)

→ how use Mutex to maximize concurrency?

Setup¶

The target function¶

HTTP GET request and reading the request body

code/go/caching/memotest/memotest.go

func httpGetBody(url string) (interface{}, error) {
    resp, err := http.Get(url)
    if err != nil {
        return nil, err
    }
    defer resp.Body.Close()
    return io.ReadAll(resp.Body) // io.ReadAll returns:
                                 // `(interface{}, error)`
}

gopl.io, p. 272:

ReadAll returns two results, a []byte and an error, but since these are assignable to the declared result types of httpGetBody—interface{} and error, respectively—we can return the result of the call without further ado. We chose this return type for httpGetBody so that it conforms to the type of functions that our cache is designed to memorize.

The data for caching¶

// Func is the type of the function to memorize
type Func func(key string) (interface{}, error)

// result stores the output of Func
type result struct {
  value interface{}
  err   error
}

// Memo caches the results of Func
type Memo struct {
  f     Func
  cache map[string]result
}

gopl.io, p. 273:

A Memo instance holds the function f to memoize, of type Func, and the cache, which is a mapping from strings to results. Each result is simply the pair of results returned by a call to f—a value and an error.

Create a new cache¶

func New(f Func) *Memo {
  return &Memo{f: f, cache: make(map[string]result)}
}

The method for caching¶

// WARNING: not concurrency-safe
func (memo *Memo) Get(key string) (interface{}, error) {
  res, ok := memo.cache[key]
  if !ok { // if the key does not exist in cache
    res.value, res.err = memo.f(key) // call the function
    memo.cache[key] = res            // store the result
  }
  return res.value, res.err
}

Example usage of Memo¶

// create new object (`memo` is the package name)
m := memo.New(httpGetBody)

// get result of the function call:
// ... is either a new http request or a cache retrieval
value, err := m.Get("https://github.com")

code: code/go/caching/memo1
run test:

cd memo1
go test -run=TestConcurrent -v . # [optional: -race]

testing code is in: code/go/caching/memotest

Concurrency-safe Memo¶

use sync.Mutex

type Memo struct {
  f     Func
  mu    sync.Mutex // guards cache variable below
  cache map[string]result
}

Concurrency-safe blocking cache¶

blocking version: code/go/caching/memo2

// concurrency-safe & blocking Get method:
  memo.mu.Lock()   // request lock
  res, ok := memo.cache[key]
  if !ok {
    res.value, res.err = memo.f(key)
    memo.cache[key] = res
  }
  memo.mu.Unlock() // release lock
  return res.value, res.err

🐢 serializes the access to entire cache
→ poor performance

gopl.io, p. 275:

By holding the lock for the duration of each call to f, Get serializes all the I/O operations we intended to parallelize. What we need is a non-blocking cache, one that does not serialize calls to the function it memoizes.

Concurrency-safe nonblocking cache¶

minimize shared memory access:
lock/unlock twice in Get: code/go/caching/memo3

// concurrency-safe & nonblocking Get method:
  memo.mu.Lock()     // request lock
  res, ok := memo.cache[key]
  memo.mu.Unlock()   // release lock right away
  if !ok {
    res.value, res.err = memo.f(key)
    memo.mu.Lock()   // request lock
    memo.cache[key] = res
    memo.mu.Unlock() // release lock
  }
  return res.value, res.err

→ causes redundant work if two goroutines get the same URL at the same time

gopl.io, p. 275:

… calling goroutine acquires the lock twice: once for the lookup, and then a second time for the update if the lookup returned nothing. In between, other goroutines are free to use the cache.

… now we notice that some URLs are being fetched twice. This happens when two or more goroutines call Get for the same URL at about the same time. Both consult the cache, find no value there, and then call the slow function f. Then both of them update the map with the result they obtained. One of the results is overwritten by the other.

Concurrency-safe nonblocking & duplicate suppression¶

add ready channel to announce that calling f completed: code/go/caching/memo4

type entry struct {
  res   result
  ready chan struct{} // closed when `res` is ready
}

type Memo struct {
  f     Func
  mu    sync.Mutex
  cache map[string]*entry // we change `result` to `entry`
}

Duplicate suppression in Get method¶

// concurrency-safe & nonblocking & duplicate suppressing Get method:
  memo.mu.Lock()
  e := memo.cache[key]
  if e == nil {         // (if this is a new request)
    e = &entry{ready: make(chan struct{})}
    memo.cache[key] = e // new cache entry, `ready` channel is open
    memo.mu.Unlock()
    e.res.value, e.res.err = memo.f(key)
    close(e.ready)      // "broadcast" ready condition
  } else {              // (otherwise it's a repeated request)
    memo.mu.Unlock()
    <-e.ready           // wait for ready condition
  }
  return e.res.value, e.res.err // safe to access `.value`, `.err`

gopl.io, p. 277:

A call to Get now involves acquiring the mutex lock that guards the cache map, looking in the map for a pointer to an existing entry, allocating and inserting a new entry if none was found, then releasing the lock. If there was an existing entry, its value is not necessarily ready yet—another goroutine could still be calling the slow function f—so the calling goroutine must wait for the entry’s “ready: condition before it reads the entry’s result. It does this by reading a value from the ready channel, since this operation blocks until the channel is closed.

If there was no existing entry, then by inserting a new “not ready” entry into the map, the current goroutine becomes responsible for invoking the slow function, updating the entry, and broadcasting the readiness of the new entry to any other goroutines that might (by then) be waiting for it.

Performance trade-offs¶

we improves:

latency
throughput

with additional costs:

memory usage
code complexity

Case study: chat server¶

Chat server¶

Idea¶

several users broadcast textual messages to each other

Need four kinds of goroutines¶

🤖 main goroutine
🔊 broadcaster (multiplexing with select)
🔔 connection handling (one for each client)
📨 writing to client (one for each client)

Main goroutine¶

listen for and accept incoming network connections from clients

func main() {
  listener, err := net.Listen("tcp", "localhost:8000") // check err
  go broadcaster()
  for {
    conn, err := listener.Accept() // check err
    go handleConn(conn)
  }
}

server: code/go/chat_server/chat
client: code/go/chat_server/netcat3

Broadcaster goroutine data¶

data for the broadcaster, in code/go/chat_server/chat

type client chan<- string // channel for outgoing messages

var (
  entering = make(chan client)
  leaving = make(chan client)
  messages = make(chan string) // all incoming client messages
)

func broadcaster() {
    clients := make(map[client]bool) // all connected clients
    for {
        // ... broadcasting select code ...
    }
}

Broadcaster function body¶

in code/go/chat_server/chat

// broadcasting select code:
select {
case msg := <-messages:
    // Broadcast incoming message to all clients' channels
    for cli := range clients {
        cli <- msg
    }

case cli := <-entering:
    clients[cli] = true

case cli := <-leaving:
    delete(clients, cli)
    close(cli)
}

gopl.io, p. 254:

The broadcaster listens on the global entering and leaving channels for announcements of arriving and departing clients. When it receives one of these events, it updates the clients set, and if the event was a departure, it closes the client’s outgoing message channel. The broadcaster also listens for events on the global messages channel, to which each client sends all its incoming messages. When the broadcaster receives one of these events, it broadcasts the message to every connected client.

Client connection handling goroutine¶

func handleConn(conn net.Conn) {
  ch := make(chan string)   // outgoing client messages
  go clientWriter(conn, ch) // writes messages in `ch` to `conn`

  // establish a new connection
  who := conn.RemoteAddr().String()
  ch <- "You are " + who
  messages <- who + " has arrived"
  entering <- ch

  // process messages
  input := bufio.NewScanner(conn)
  for input.Scan() {
    messages <- who + ": " + input.Text()
  }
  leaving <- ch
  messages <- who + " has left"
  conn.Close()
}

func clientWriter(conn net.Conn, ch <-chan string) {
  for msg := range ch {
    fmt.Fprintln(conn, msg)
  }
}

Key takeaways¶

concurrency → faster workloads and overcoming I/O bottlenecks
patterns matter: caching & broadcasting
trade-offs: speed vs complexity
rule of thumb: channels for coordination, mutexes for accessing one shared state

Looking ahead¶

scalability: performance vs resources
how well does my program scale, really?

Prepare for the next session 🌱¶

sessions/16-todo