Broadcasting to Many Clients
One goroutine owns the set of clients and takes register, unregister and broadcast over channels, so no mutex is needed.
select { case c := <-h.register: ...; case c := <-h.unregister: ...; case msg := <-h.broadcast: ... }One message often has to reach every connected client, and the set of clients changes every second as people arrive and leave. That is shared state, touched by many goroutines at once.
There are two ways to make it safe. One is a lock: a map of clients with a mutex around it. The other is to give the map to a single goroutine and have every other goroutine ask that goroutine to do the work. The second goroutine is called a hub, and this lesson uses it because the client set is written far more often than it is read.
How a hub works
The hub owns three channels: register, unregister and broadcast. Its whole body is a select over those three. A new connection sends itself on register, a finished one sends itself on unregister, and a producer sends bytes on broadcast, at which point the hub walks its map and passes the bytes to each client. Because one goroutine does all of it, the map needs no mutex, and every change to the client set happens in one readable place.
Slow clients
Each client has its own buffered channel with its own writing goroutine draining it. The hub sends with a select that has a default case, so the send never blocks. A full buffer means that client cannot keep up, and the hub drops it. Without that, one person on a bad connection stops messages for everyone.
What trips people up
Only the hub closes a client's channel, and it deletes the client in the same step. Delete on unregister, and delete an empty room too, or the map grows for the life of the process.
Example
package main
import "fmt"
// client is one connected browser. send carries messages out towards it.
type client struct {
name string
send chan string
}
// hub owns the set of clients. Only run touches that map, so no mutex appears
// anywhere in this file.
type hub struct {
register chan *client
unregister chan *client
broadcast chan string
clients map[*client]bool
}
func (h *hub) run() {
for {
select {
case c := <-h.register:
h.clients[c] = true
case c := <-h.unregister:
if h.clients[c] {
delete(h.clients, c)
close(c.send) // tells that client's writer goroutine to stop
}
case msg := <-h.broadcast:
for c := range h.clients {
select {
case c.send <- msg:
default: // this one is behind: drop it, do not stall the rest
delete(h.clients, c)
close(c.send)
}
}
}
}
}
func main() {
h := &hub{
register: make(chan *client),
unregister: make(chan *client),
broadcast: make(chan string),
clients: make(map[*client]bool),
}
go h.run()
alice := &client{name: "alice", send: make(chan string, 4)}
bob := &client{name: "bob", send: make(chan string, 4)}
h.register <- alice
h.register <- bob
h.broadcast <- "price: 42"
for _, c := range []*client{alice, bob} {
fmt.Printf("%s received %q\n", c.name, <-c.send)
}
h.unregister <- bob
h.broadcast <- "price: 43"
fmt.Printf("%s received %q\n", alice.name, <-alice.send)
if _, open := <-bob.send; !open {
fmt.Println("bob's channel is closed")
}
}
// Output:
// alice received "price: 42"
// bob received "price: 42"
// alice received "price: 43"
// bob's channel is closed
When to use it
- Deliver one chat message to every member of a room.
- Push a price change to every dashboard that is currently open.
- Disconnect a client whose connection is too slow to keep up with the stream.
More examples
A hub with rooms
The same hub with one set of clients per room, which is what a chat service or a per-tenant dashboard needs. Keeping drop as a method makes the rule obvious: it is called only from run, so it needs no lock, and it removes the client, closes its channel and forgets the room in one step. Deleting the empty room matters in services where rooms are created from user input, because otherwise the map only ever grows.
package chat
// message is one broadcast: a room name and the bytes to send.
type message struct {
room string
data []byte
}
type client struct {
room string
send chan []byte
}
// hub keeps one set of clients per room. Nothing outside run touches rooms.
type hub struct {
register chan *client
unregister chan *client
broadcast chan message
rooms map[string]map[*client]bool
}
func newHub() *hub {
return &hub{
register: make(chan *client),
unregister: make(chan *client),
broadcast: make(chan message),
rooms: make(map[string]map[*client]bool),
}
}
func (h *hub) run() {
for {
select {
case c := <-h.register:
if h.rooms[c.room] == nil {
h.rooms[c.room] = make(map[*client]bool)
}
h.rooms[c.room][c] = true
case c := <-h.unregister:
h.drop(c)
case m := <-h.broadcast:
for c := range h.rooms[m.room] {
select {
case c.send <- m.data:
default:
h.drop(c) // too slow: disconnect it rather than stall the room
}
}
}
}
}
// drop is called only from run, so it needs no lock.
func (h *hub) drop(c *client) {
if _, ok := h.rooms[c.room][c]; !ok {
return
}
delete(h.rooms[c.room], c)
close(c.send)
if len(h.rooms[c.room]) == 0 {
delete(h.rooms, c.room) // an empty room is a slow leak
}
}Joining and leaving from the handler
Where a connection meets the hub. The handler upgrades, registers itself, starts its single writer goroutine, and then becomes the reader itself, which is why it must not return early. The deferred unregister is the only cleanup path, and it runs whether the read loop ended on a close frame, a read deadline or a broken network. The buffer of sixteen messages is the slack a client gets before the hub gives up on it.
package chat
import (
"net/http"
"github.com/gorilla/websocket"
)
type client struct {
ws *websocket.Conn
send chan []byte
}
type hub struct {
register chan *client
unregister chan *client
}
var upgrader = websocket.Upgrader{}
// serveWS is where a connection joins the hub and where it leaves again. The
// handler itself becomes the reading goroutine, so it must not return early.
func (h *hub) serveWS(w http.ResponseWriter, r *http.Request) {
ws, err := upgrader.Upgrade(w, r, nil)
if err != nil {
return
}
c := &client{ws: ws, send: make(chan []byte, 16)} // 16 messages of slack
h.register <- c
go c.writePump() // one writer for this connection
defer func() {
h.unregister <- c // the hub closes c.send, which stops writePump
ws.Close()
}()
for {
if _, _, err := ws.ReadMessage(); err != nil {
return
}
}
}
func (c *client) writePump() {
defer c.ws.Close()
for data := range c.send {
if err := c.ws.WriteMessage(websocket.TextMessage, data); err != nil {
return
}
}
}Shutting the hub down without leaking
A hub that runs forever is fine until the process needs to exit. Taking a context lets graceful shutdown stop it, and the deferred loop closes every client channel so that each writer goroutine ends instead of blocking on a channel nobody will ever send to again. Pair this with the server shutdown in your main function so a rolling deploy does not leave goroutines waiting.
package chat
import (
"context"
"log"
)
type client struct {
send chan string
}
type hub struct {
register chan *client
broadcast chan string
clients map[*client]bool
}
// run stops when ctx is cancelled, and closes every client on the way out so
// that each writer goroutine ends instead of leaking.
func (h *hub) run(ctx context.Context) {
defer func() {
for c := range h.clients {
delete(h.clients, c)
close(c.send)
}
log.Print("hub stopped")
}()
for {
select {
case <-ctx.Done():
return
case c := <-h.register:
h.clients[c] = true
case msg := <-h.broadcast:
for c := range h.clients {
select {
case c.send <- msg:
default:
delete(h.clients, c)
close(c.send)
}
}
}
}
}
Discussion