3
\$\begingroup\$

I've never done concurrent programming before, but I've written one using go channels and RWMutex, but I'm not sure if my approach is idiomatic. I feel like it could all use channels (Maybe).

The protocol is simple: Messages are delimited by '\n'. The first message client sends is it's name. The rest are normal messages. I used a map because it's add + delete is faster, and because I'm not concerned about the ordering. I'm mostly concerned with the locks at the last 4 methods. Could I use channels here instead? Am I using locks correctly?

type message struct {
    message string
    client  *client
}

type client struct {
    conn   net.Conn
    reader *bufio.Reader
    name   string
}

type Server struct {
    l           net.Listener
    clients     map[*client]struct{}
    messageChan chan message
    port        uint16
    name        string
    m           sync.RWMutex
}

// NewServer starts listening, and returns an initialized server. If an error occured while listening started, (nil, error) is returned.
func NewServer(name string, port uint16) (*Server, error) {
    p := strconv.FormatUint(uint64(port), 10)
    l, e := net.Listen("tcp", "localhost:"+p)
    if e != nil {
        return nil, e
    }
    return &Server{l, make(map[*client]struct{}), make(chan message), port, name, sync.RWMutex{}}, nil
}

// Start starts accepting clients + managing messages.
func (r *Server) Start() {
    go r.sendMessages()
    for {
        conn, e := r.l.Accept()
        if e != nil {
            continue
        }
        go r.handleConn(conn)
    }
}

func (r *Server) handleConn(conn net.Conn) {
    reader := bufio.NewReader(conn)
    name, e := reader.ReadString('\n')
    name = strings.TrimSpace(name)
    if e != nil {
        conn.Close()
        return
    }

    c := &client{conn, reader, name}
    r.addClient(c)
    r.publishMessage(c.name + " has joined the server\n")
    r.handleClient(c)
}

func (r *Server) handleClient(c *client) {
    for {
        msg, e := c.reader.ReadString('\n')
        if e == io.EOF {
            c.conn.Close()
            r.deleteClient(c)
            r.publishMessage(c.name + " has left the server\n")
            return
        } else if e != nil {
            continue
        }
        r.messageChan <- message{c.name + ": " + msg, c}
    }
}

func (r *Server) sendMessages() {
    for {
        m := <-r.messageChan
        r.m.RLock()
        for k := range r.clients {
            if k != m.client {
                go k.conn.Write([]byte(m.message))
            }
        }
        r.m.RUnlock()
        fmt.Print(m.message)
    }
}

func (r *Server) publishMessage(msg string) {
    r.m.RLock()
    for k := range r.clients {
        go k.conn.Write([]byte(msg))
    }
    r.m.RUnlock()
    fmt.Print(msg)
}

func (r *Server) addClient(c *client) {
    r.m.Lock()
    r.clients[c] = struct{}{}
    r.m.Unlock()
}

func (r *Server) deleteClient(c *client) {
    r.m.Lock()
    delete(r.clients, c)
    r.m.Unlock()
}

I would appreciate if anyone could give some suggestions/feedback.

\$\endgroup\$
4
\$\begingroup\$

This is a nice project to learn concurrency on, and you've already identified a few issues that concern you, and you're right in some ways. Go documentation recommends using channels when possible, over mutexes. The documentation in the sync package says: "Other than the Once and WaitGroup types, most are intended for use by low-level library routines. Higher-level synchronization is better done via channels and communication."

Channels are your friend in Go, and you should use them liberally (well, more liberally than mutexes).

There are some special cases with channels, though, and your code has a good example of how they can be used in a broken way. Your concurrency is not nearly as significant as you think it is. I will focus on this code for the moment, and we'll simplify it down, and then get back to the concurrency problem in a bit...

Client handler

This code....

func (r *Server) handleConn(conn net.Conn) {
    reader := bufio.NewReader(conn)
    name, e := reader.ReadString('\n')
    name = strings.TrimSpace(name)
    if e != nil {
        conn.Close()
        return
    }

    c := &client{conn, reader, name}
    r.addClient(c)
    r.publishMessage(c.name + " has joined the server\n")
    r.handleClient(c)
}

func (r *Server) handleClient(c *client) {
    for {
        msg, e := c.reader.ReadString('\n')
        if e == io.EOF {
            c.conn.Close()
            r.deleteClient(c)
            r.publishMessage(c.name + " has left the server\n")
            return
        } else if e != nil {
            continue
        }
        r.messageChan <- message{c.name + ": " + msg, c}
    }
}

... takes a client network connection, pulls the name, and then establishes a loop for reading and distributing messages. It distributes the messages by pushing them on to the r.messageChan channel. Let's simplify that code a whole bunch, and point out some problems as we go.

First up, you need to learn the defer statement, and use it for closing the client connection. If you change the first line of the function to be:

func (r *Server) handleConn(conn net.Conn) {
    defer conn.Close()

now you don't need to worry about closing it in a number of other places. Just returning from the function is enough to close it.

Secondly, you should use a scanner... not a Reader. See: Scanner which has the documentation "Scanner provides a convenient interface for reading data such as a file of newline-delimited lines of text." Scanner defaults to using line-termination (which is what you have) so it's easy to set up. Your code:

    reader := bufio.NewReader(conn)

becomes:

     scanner := bufio.NewScanner(conn)

Note that Scanner never returns an EOF error, it just returns a false scan at EOF (or any other error). The advantage for you of a scanner is that it simplifies your loops, and error handling. This is not obvious immediately in your case as you scan the name, but will become apparent when processing messages. To get the name, you need:

// find the next end-of-line.. (or perhaps EOF)
if !scanner.Scan() {
    // could not scan anything on the first scan... odd... no name.
    // the scanner.Err() may have a value if there's a problem, but we don't care.
    return
}
name := strings.TrimSpace(scanner.Text())

OK, that's how you get the name.... and now we can create the client (using the scanner instead of reader) and we also use a defer function to remove the client instead of shifting that logic in to some other place (keeping the clean-up code in the same place as the mess-up code is important for maintainability). Also, we should relocate the "publish" methods in to the addClient and deleteClient methods (where they really should be anyway):

c := &client{conn, scanner, name}
r.addClient(c)
defer r.deleteClient(c)
r.handleClient(c)

OK, so now we have a client that can process incoming messages, let's look at that code, but using a scanner now instead of a reader, and it does not need to do the deleteClient or the related "publish" of the deletion. It also does not need to close the connection:

func (r *Server) handleClient(c *client) {
    for c.scanner.Scan() {
        msg := c.scanner.Text()
        r.messageChan <- message{c.name + ": " + msg, c}
    }
    // we could check scanner.Err() here, but we are ignoring all non-EOF errors anyway
}

Now, that looks neat... so small now that it's hardly worth having it's own function, let's pull it back in to the handleConn function (which we will rename to handleClient because I prefer client... and we'll remove the scanner from the client struct because no-one else uses it in any other place... our handleClient method is now:

func (r *Server) handleClient(conn net.Conn) {
    defer conn.Close()

    scanner := bufio.NewScanner(conn)

    // read off the first line as the "name"

    // find the next end-of-line.. (or perhaps EOF)
    if !scanner.Scan() {
        // could not scan anything on the first scan... odd... no name.
        // the scanner.Err() may have a value if there's a problem, but we don't care.
        return
    }
    name := strings.TrimSpace(scanner.Text())

    // register our client in to the control systems.
    c := &client{conn, name}
    r.addClient(c)
    defer r.deleteClient(c)

    // send messages in to the channel.
    for scanner.Scan() {
        r.messageChan <- message{name + ": " + scanner.Text(), c}
    }
}

Now, that's a bunch simpler, especially the reduced error handling, and the reduced content in the client.

There were a few things to hammer out here. We shifted the printing of messages like X has joined the server and X has left the server in to the addClient and deleteClient methods respectively. This reduces the mixed logic of handling a client and also reporting activity in this function, and shifts it to the more logical place where we are actually adding and deleting things. We used defer functions to handle cleanup of resources at the same time/place that the resources were created. We used a scanner instead of a reader.

messageChan

Let's cover that concurrency issue I mentioned earlier. It comes down to the messageChan. The problem is only one client is busy at a time. Let me explain... this code here:

r.messageChan <- message{name + ": " + scanner.Text(), c}

pushes messages on to the channel, and this code pulls them off:

for {
    m := <-r.messageChan
    .... distribute the messages to other clients
}

You would think that multiple clients could all be pushing messages, and the for-loop is pulling them and distributing them at the same time. Unfortunately, all the clients "block" at the point where they push the message, and they all wait there, until the for-loop starts, and it will pull just one message. That one message will be processed, and the client that happened to have its message pulled, will then go off and wait for another message - all the other clients are still blocked. Once the for loop has sent the message off, it will come back and read another from the channel, and one other client will be lucky enough to have its message delivered.

This is because you have created an unbuffered channel. The documentation has this to say: "If the capacity is zero or absent, the channel is unbuffered and communication succeeds only when both a sender and receiver are ready." Since the receiver is only ready once each loop, the receiver essentially serializes all the clients.

You need to declare your channel with some buffer size if you want to avoid this issue. Even a size of 1 would be beneficial. A larger buffer may result in caching of messages but that should not be a problem.

Communication instead of Mutexes

Communication in Go is an odd concept to explain. The purpose of mutexes is to allow safe access to resources from multiple go-routines. Communication is sort of the opposite - communication basically implies only one go-routine ever accesses a resource, so mutexes are not needed, but we can communicate the changes (normally via a channel) to that go-routine.

If only one routine accesses a resource, no mutex is needed. Let's see how we can apply that to your code. Your "server" code really has three things going on...

  1. clients get added
  2. messages get distributed
  3. clients get removed.

You've already got a channel for the messages, so let's look at the other parts....

func (r *Server) addClient(c *client) {
    r.m.Lock()
    r.clients[c] = struct{}{}
    r.m.Unlock()
}

func (r *Server) deleteClient(c *client) {
    r.m.Lock()
    delete(r.clients, c)
    r.m.Unlock()
}

Let's turn those in to channels instead:

func (r *Server) addClient(c *client) {
    r.added <- c
}

func (r *Server) deleteClient(c *client) {
    r.deleted <- c
}

We will have to create those channels (with some buffering) on the Server... but, that's how we communicate the add/remove from the client.

How do we handle that on the server? Currently the server code looks like:

func (r *Server) sendMessages() {
    for {
        m := <-r.messageChan
        r.m.RLock()
        for k := range r.clients {
            if k != m.client {
                go k.conn.Write([]byte(m.message))
            }
        }
        r.m.RUnlock()
        fmt.Print(m.message)
    }
}

If we listen for communication from all channels, not just the messageChan, we can do it all in that routine (and no need for mutexes because we're the only ones to manipulate the client map.

func (r *Server) sendMessages() {
    for {
        select {
        case m := <-r.messageChan:
            for k := range r.clients {
                if k != m.client {
                    go k.conn.Write([]byte(m.message))
                }
            }
            fmt.Print(m.message)
        case c := <-r.added:
            r.publishMessage(c.name + " has joined the server\n")
            r.clients[c] = struct{}{}
        case c := <-r.deleted:
            r.publishMessage(c.name + " has left the server\n")
            delete(r.clients, c)
        }
    }
}

Now we can even remove the client map from the Server completely, and keep it local inside thesendMessages` go-routine.

Read up on the select statement which is how communication from multiple sources is handled: https://golang.org/ref/spec#Select_statements

Conclusion

This post has been long enough already... hopefully you'll get something out of it. I hope that you at least get how the unbuffered channels can impact things, and that communication with select-statements allows you to localize resources to just one routine, meaning mutexes are not needed.

\$\endgroup\$
  • \$\begingroup\$ Thank you so much for the excellent and thorough answer. In the future, I'll try and use channels and avoid sharing memory whenever possible. What if two users join at the same time and send the messages at the same time while select cases have not run yet (Because they are buffered channels)? I don't think it is likely to occur in practice, but still possible. Wouldn't that make the select process the messages in random order (In the worst case, message being sent before client is added). How would I make sure it's in order? \$\endgroup\$ – Winger Sendon Feb 16 '17 at 12:53
  • 1
    \$\begingroup\$ About the order, in each channel the insertion order is a first-come-first-served process, so the order of retrieval depends on that. Note that there is no way to predict which routine succeeds first if multiple are all blocked and waiting for space on the channel (which is a true statement even for your original code - you can't tell which client posts the message first). On the other hand, a larger channel size reduces the possibility that multiple routines are all waiting. \$\endgroup\$ – rolfl Feb 16 '17 at 13:02
  • 1
    \$\begingroup\$ Note that an arriving message for a channel that has not yet been "added" will be distributed to all clients except that client (which is normal ;-) and sending a message to a client that's been closed, but not yet deleted, will have an IO error on the write, but you ignore those errors anyway. Nothing bad will happen with out-of-order communication in your code. If you need strict ordering then have a single channel with a "message type" type of field. \$\endgroup\$ – rolfl Feb 16 '17 at 13:02

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.