Concurrent Servers: Part 8 - Go - Eli Bendersky's website

Go's runtime multiplexes lightweight goroutines onto OS threads with its own M:N scheduler, so the language sidesteps most of the machinery the earlier parts of this series had to build by hand. This part follows the same progression as the prior ones — sequential server first, then concurrency, then ways to bound it — using the state machine protocol from part 1 and the primality tester from part 4.

Starting sequential

As in earlier installments, the baseline is an infinite server: it accepts connections until it is explicitly killed. The main function wires up the listener, while a per-client function carries out the protocol on a net.Conn that represents the socket to the connected client.

func main() {
  port := "9090"
  if len(os.Args) >= 2 {
    port = os.Args[1]
  }
  log.Println("Serving on port", port)

  listener, err := net.Listen("tcp", ":"+port)
  if err != nil {
    log.Fatal("Error listening:", err)
  }
  defer listener.Close()

  for {
    conn, err := listener.Accept()
    if err != nil {
      log.Printf("Error accepting connection: %v", err)
      continue
    }
    log.Println("connection received from", conn.RemoteAddr())

    if err := server.ServeSerialProtocol(conn); err != nil {
      log.Printf("Error serving %v: %v", conn.RemoteAddr(), err)
    } else {
      log.Println("peer done", conn.RemoteAddr())
    }
  }
}

type processingState int

const (
  waitForMsg processingState = iota
  inMsg
)

// ServeSerialProtocol serves our serial protocol to a single TCP connection.
func ServeSerialProtocol(conn net.Conn) error {
  defer conn.Close()

  if _, err := conn.Write([]byte{'*'}); err != nil {
    return err
  }

  var state processingState = waitForMsg
  buf := make([]byte, 1024)
  for {
    n, err := conn.Read(buf)
    for _, b := range buf[:n] {
      switch state {
      case waitForMsg:
        if b == '^' {
          state = inMsg
        }
      case inMsg:
        if b == '$' {
          state = waitForMsg
        } else {
          var bb byte = byte(b) + 1
          if _, err := conn.Write([]byte{bb}); err != nil {
            return err
          }
        }
      }
    }

    // Check error after processing the bytes received along with it.
    if err != nil {
      if errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) {
        return nil
      } else {
        return err
      }
    }
  }
}

One goroutine per client

Concurrency begins by launching a goroutine for each accepted connection; only the code that differs from the sequential sample is highlighted.

func main() {
  port := "9090"
  if len(os.Args) >= 2 {
    port = os.Args[1]
  }
  log.Println("Serving on port", port)

  listener, err := net.Listen("tcp", ":"+port)
  if err != nil {
    log.Fatal("Error listening:", err)
  }
  defer listener.Close()

  for {
    conn, err := listener.Accept()
    if err != nil {
      log.Printf("Error accepting connection: %v", err)
      continue
    }
    log.Println("connection received from", conn.RemoteAddr())

    go func() {
      if err := server.ServeSerialProtocol(conn); err != nil {
        log.Printf("Error serving %v: %v", conn.RemoteAddr(), err)
      } else {
        log.Println("peer done", conn.RemoteAddr())
      }
    }()
  }
}

The change is small because the server never terminates, so nothing needs to join these goroutines — there is no role for a sync.WaitGroup here. Arguments to server.ServeSerialProtocol come from the enclosing scope via lexical capture, and the surrounding closure takes care of the return value.

Goroutines are cheap enough that resource exhaustion from spawning too many of them is unlikely; file descriptors for sockets will typically run out first. Even so, capping concurrency remains useful in certain situations, which the next sections cover.

Bounding concurrency

Several conditions argue for limiting parallel work despite the low cost of goroutines:

• Compute-heavy tasks share a fixed CPU budget. When many goroutines contend for few cores, all of them crawl; fewer tasks that finish in reasonable time can be preferable.
• Downstream resources are often rate-limited or scarce — database connections, calls to other services. If each task talks to such a service, its concurrency has to be carefully managed.
• Work dictated by clients is a security concern: a malicious client can overload or crash a service that is too eager to serve, denying legitimate users.

For the remainder, the example shifts to the primality testing server from part 4 — numbers arrive, a sleep simulates blocking work, and the reply is "prime" or "composite". The unbounded one-goroutine-per-client variant is nearly the same as the previous sample, differing only in which function the goroutine invokes:

go func() {
  if err := server.ServePrimeProtocol(conn); err != nil {
    log.Printf("Error serving %v: %v", conn.RemoteAddr(), err)
  } else {
    log.Println("peer done", conn.RemoteAddr())
  }
}()

with ServePrimeProtocol defined as [1]:

// ServePrimeProtocol serves our prime protocol to a single TCP connection.
func ServePrimeProtocol(conn net.Conn) error {
  defer conn.Close()
  buf := make([]byte, 1024)

  for {
    n, readerr := conn.Read(buf)
    if n > 0 {
      // Parse the read buffer to an i64
      num, err := strconv.ParseInt(strings.TrimSpace(string(buf[:n])), 10, 64)
      if err != nil {
        return err
      }

      response := "composite"
      if isPrime(num, true) {
        response = "prime"
      }
      if _, err := conn.Write([]byte(response + "\n")); err != nil {
        return err
      }
    }

    if readerr != nil {
      if errors.Is(readerr, io.EOF) || errors.Is(readerr, net.ErrClosed) {
        return nil
      }
      return readerr
    }
  }
}

// isPrime returns true if n is prime, false otherwise. If delay is true, it
// will sleep for n milliseconds before calculating.
func isPrime(n int64, delay bool) bool {
  if delay {
    time.Sleep(time.Duration(n) * time.Millisecond)
  }

  if n < 2 {
    return false
  }
  if n%2 == 0 {
    return n == 2
  }
  for i := int64(3); i*i <= n; i += 2 {
    if n%i == 0 {
      return false
    }
  }
  return true
}

A counting semaphore is the most direct way to cap concurrency in Go, and a channel provides it:

func main() {
  port := "8070"
  if len(os.Args) >= 2 {
    port = os.Args[1]
  }
  log.Println("Serving on port", port)

  listener, err := net.Listen("tcp", ":"+port)
  if err != nil {
    log.Fatal("Error listening:", err)
  }
  defer listener.Close()

  maxConcurrency := runtime.NumCPU()
  sem := make(chan struct{}, maxConcurrency)

  for {
    conn, err := listener.Accept()
    if err != nil {
      log.Printf("Error accepting connection: %v", err)
      continue
    }
    log.Println("connection received from", conn.RemoteAddr())

    // Acquire a token from the semaphore to limit concurrency.
    sem <- struct{}{}

    go func() {
      // Return the token when done serving the connection.
      defer func() { <-sem }()

      if err := server.ServePrimeProtocol(conn); err != nil {
        log.Printf("Error serving %v: %v", conn.RemoteAddr(), err)
      } else {
        log.Println("peer done", conn.RemoteAddr())
      }
    }()
  }
}

Here the channel sem acts as the semaphore, and it must be bounded with a maximum size. Sending to it acquires a token; receiving from it releases one. Once the channel is full, sem <- struct{}{} blocks until another goroutine removes a token [2]. The element type is struct{} — an empty value — the idiomatic choice for a channel used purely for its signaling semantics rather than for carrying data.

A worker pool, when it earns its keep

Given how cheap goroutine creation is and how little limiting costs, the worker pool pattern is often superfluous for servers like this one. It still has niches — workers that must carry non-trivial state from task to task, for instance — so it is worth showing.

func worker(jobs <-chan net.Conn) {
  for conn := range jobs {
    if err := server.ServePrimeProtocol(conn); err != nil {
      log.Printf("Error serving %v: %v", conn.RemoteAddr(), err)
    } else {
      log.Println("peer done", conn.RemoteAddr())
    }
  }
}

func main() {
  port := "8070"
  if len(os.Args) >= 2 {
    port = os.Args[1]
  }
  log.Println("Serving on port", port)

  listener, err := net.Listen("tcp", ":"+port)
  if err != nil {
    log.Fatal("Error listening:", err)
  }
  defer listener.Close()

  // The channel is unbuffered, so it will block when there are no available
  // workers to accept a new connection.
  jobs := make(chan net.Conn)
  maxConcurrency := runtime.NumCPU()
  for i := 0; i < maxConcurrency; i++ {
    go worker(jobs)
  }

  for {
    conn, err := listener.Accept()
    if err != nil {
      log.Printf("Error accepting connection: %v", err)
      continue
    }
    log.Println("connection received from", conn.RemoteAddr())

    jobs <- conn
  }
}

A fixed set of worker goroutines is started, and all of them pull jobs from one shared channel. Each accepted client connection becomes a job on that channel inside the Accept loop, and the next free worker picks it up. Clean shutdown would normally involve a sync.WaitGroup, but since this server never exits, that bookkeeping is unnecessary.

Is async programming needed?

Practically never, in my experience. Go was built with large-scale concurrency as a first-class goal: goroutines start cheaply, occupy a tiny memory footprint, and context switches happen fast and entirely in user space. Measurements from 2018 put switching time at roughly 170 ns, against 1-2 microseconds for Linux threads.

The runtime also sits on top of event-driven I/O facilities such as epoll already. A goroutine blocked on a socket is effectively parked and consumes nothing beyond its small memory footprint; the runtime wakes it when its descriptor is ready — structurally very close to how asynchronous programming works.

That said, some practitioners push further and program directly against async APIs when handling millions of concurrent streams. This is a rare case, and the overwhelming majority of users never need it.

Closing thoughts

A 2018 post of mine argued that Go hits the concurrency nail right on the head, and after several more years of active coding I stand by it. While other environments invest heavily in async-await style event loops implemented in libraries, Go bakes event-driven I/O into the language and runtime: lightweight green threads that can also run blocking work, without dragging users into the function coloring problem.

Code

All the code for this post is available on GitHub.