golang concurrent tcp server

Question

Go and concurrency noob here. I wrote a program to handle a tcp connection to index (and removes) packages. I want it to be able hundreds of connections and messages that might come in at the same time. I am using a hash table to store the data and using channels to provide the worker with the incoming messages.

My understanding is that I will not have to use any locks against the data store as all add/remove will be blocked until the previous connection has finished writing and reading. Just wanted to make sure I understand go channels correctly here and I wrote an efficient solution. In the future in lieu of the hash table does it make sense to use something like Redis? Or even store a list on the file system?

I would like to understand

• My use of channels is correct and additional locks won't be necessary. In the future the comments will be a list of other package dependencies so it's important each request can write to the data correctly
• I'm using a hash table as the data store but since this is in memory eventually would it make sense to use something like Redis? Or even writing to a local file
• How would one test sending hundreds of messages to this server?
• I do check for the format of the message but what happens if a client disconnects abruptly?

Usage of the app:

go run main.go
# in a separate terminal
echo -n "INDEX|vim|comment\n" | nc localhost 3333
echo -n "REMOVE|vim|blah,blah\n" | nc localhost 3333
echo -n "INDEX|vim|comment\n" | nc localhost 3333
echo -n "INDEX|vim|comment\n" | nc localhost 3333
echo -n "COUNT|vim|comment\n" | nc localhost 3333 

The code

package main

import (
    "bufio"
    "errors"
    "fmt"
    "log"
    "net"
    "os"
    "strings"
)

const (
    CONN_HOST      = "localhost"
    CONN_PORT      = "3333"
    CONN_TYPE      = "tcp"
    INDEX_COMMAND  = "INDEX"
    REMOVE_COMMAND = "REMOVE"
    COUNT_COMMAND  = "COUNT"
)

func main() {
    l, err := net.Listen(CONN_TYPE, CONN_HOST+":"+CONN_PORT)
    if err != nil {
        fmt.Println("Error listening:", err.Error())
        os.Exit(1)
    }
    defer l.Close()
    fmt.Println("Listening on " + CONN_HOST + ":" + CONN_PORT)

    msgCh := make(chan string)
    resultCh := make(chan string)

    ds := newDataStore()
    go ds.msgHandlerWorker(msgCh, resultCh)

    for {
        conn, err := l.Accept()
        if err != nil {
            fmt.Println("Error accepting: ", err.Error())
            os.Exit(1)
        }
        go handleRequest(conn, msgCh, resultCh)
    }
}

type Message struct {
    command string
    pkg     string
    comment string
}

type DataStore struct {
    pkgInfo map[string]string
    pkgRef  map[string]uint
}

func newDataStore() *DataStore {
    return &DataStore{
        pkgInfo: make(map[string]string),
        pkgRef:  make(map[string]uint),
    }
}

func (ds *DataStore) parseMsg(msg string) (Message, error) {
    var parsedMsg Message
    m := strings.TrimSpace(msg)
    s := strings.Split(m, "|")

    if len(s) < 3 {
        return parsedMsg, errors.New("Not correct format")
    }

    parsedMsg.command = s[0]
    parsedMsg.pkg = s[1]
    parsedMsg.comment = s[2]

    return parsedMsg, nil
}

func (ds *DataStore) addToHashTable(msg Message) error {
    ds.pkgInfo[msg.pkg] = msg.comment

    if _, ok := ds.pkgRef[msg.pkg]; ok {
        ds.pkgRef[msg.pkg] += 1
    } else {
        ds.pkgRef[msg.pkg] = 0
    }

    return nil
}

func (ds *DataStore) removeFromHashTable(msg Message) error {
    delete(ds.pkgInfo, msg.pkg)

    if _, ok := ds.pkgRef[msg.pkg]; ok {
        ds.pkgRef[msg.pkg] -= 1
    }

    return nil
}

func (ds *DataStore) msgHandlerWorker(msgCh chan string, resultCh chan string) {
    for {
        msg := <-msgCh
        parsedMsg, err := ds.parseMsg(msg)
        if err != nil {
            log.Print("pailed to parse msg")
            resultCh <- "FAIL\n"
            continue
        }

        switch parsedMsg.command {
        case INDEX_COMMAND:
            if err := ds.addToHashTable(parsedMsg); err != nil {
                log.Print("Failed To add")
                resultCh <- "FAIL\n"
                continue
            }

            resultCh <- "OK\n"

        case REMOVE_COMMAND:
            if err := ds.removeFromHashTable(parsedMsg); err != nil {
                log.Print("Failed To add")
                resultCh <- "FAIL\n"
                continue
            }

            resultCh <- "OK\n"
        case COUNT_COMMAND:
            if val, ok := ds.pkgRef[parsedMsg.pkg]; ok {
                resultCh <- fmt.Sprintf("%d\n", val)
                continue
            }

            resultCh <- "FAIL\n"
        default:
            log.Print("got invalid command")
            resultCh <- "FAIL\n"
            continue
        }
    }
}

func handleRequest(conn net.Conn, msgCh chan string, resultCh chan string) {
    msg, err := bufio.NewReader(conn).ReadString('\n')
    log.Printf("msg %v", msg)
    if err != nil {
        fmt.Println("Error reading:", err.Error())
    }

    msgCh <- msg
    response := <-resultCh

    conn.Write([]byte(response))
    conn.Close()
}

Thanks for reading!

Answer 1

Before addressing your concern, a few things on coding style:

Coding style

This can easily be detected with tools like golint and go vet

• In go, always use camelCase for variables and constant names, so CONN_HOST should be connHost, or just simply host

• you can directly increment a var stored in a map, so instead of ds.pkgRef[msg.pkg] += 1, simply use ds.pkgRef[msg.pkg]++

• use fmt.Printf("Error listening: %v\n", err) instead of fmt.Println("Error listening:", err.Error())

• a more idiomatic way of creating a message would be

this

parsedMsg := Message{
    command: s[0],
    pkg:     s[1],
    comment: s[2],
}

instead of

var parsedMsg Message
parsedMsg.command = s[0]
parsedMsg.pkg = s[1]
parsedMsg.comment = s[2]

• some methods always return nil error (ie addToHashTable()), so no need to return/check for an error when calling this method

• in msgHandlerWorker(), you do not log any errors that could occur :

for example,

if err != nil {
    log.Print("pailed to parse msg")
    resultCh <- "FAIL\n"
    continue
}

should be

if err != nil {
    log.Printf("pailed to parse msg: %v\n", err)
    resultCh <- "FAIL\n"
    continue
}

---

Concurrency

Maps are not safe for concurrent use, for more details see map documentation

Here you're modifing the map from multiple goroutines, so two solutiions to fix this:

• use a sync.RWMutex
• read/write to the map from a single goroutine

I'll go for the second method as there is a simple way to handle all connection in the main goroutine:

instead of go handleRequest(...), let's handle incoming connection synchronously. There's no need for channels here, we can rewrite it like this:

for {
    conn, err := l.Accept()
    if err != nil {
        fmt.Printf("Error accepting: %v\n", err)
        os.Exit(1)
    }
    // read the incoming message 
    msg, err := bufio.NewReader(conn).ReadString('\n')
    if err != nil {
        fmt.Printf("Error reading: %v\n", err)
    }
    // directly parse the message 
    response, err := ds.handleMsg(msg)
    if err != nil {
        fmt.Printf("fail to handle message: %v\n", err)
    }
    // send the response back 
    conn.Write([]byte(response + "\n"))
    conn.Close()
}

The final code would look like:

package main

import (
    "bufio"
    "fmt"
    "net"
    "os"
    "strings"
)

const (
    host     = "localhost"
    port     = "3333"
    adress   = host + ":" + port
    connType = "tcp"
    index    = "INDEX"
    remove   = "REMOVE"
    count    = "COUNT"
)

func main() {
    l, err := net.Listen(connType, adress)
    if err != nil {
        fmt.Printf("Error listening: %v", err)
        os.Exit(1)
    }
    defer l.Close()
    fmt.Println("Listening on " + adress)

    ds := newDataStore()

    for {
        conn, err := l.Accept()
        if err != nil {
            fmt.Printf("Error accepting: %v\n", err)
            os.Exit(1)
        }
        msg, err := bufio.NewReader(conn).ReadString('\n')
        if err != nil {
            fmt.Printf("Error reading: %v\n", err)
        }
        response, err := ds.handleMsg(msg)
        if err != nil {
            fmt.Printf("fail to handle message: %v\n", err)
        }
        conn.Write([]byte(response + "\n"))
        conn.Close()
    }
}

type Message struct {
    command string
    pkg     string
    comment string
}

type DataStore struct {
    pkgInfo map[string]string
    pkgRef  map[string]uint
}

func newDataStore() *DataStore {
    return &DataStore{
        pkgInfo: make(map[string]string),
        pkgRef:  make(map[string]uint),
    }
}

func (ds *DataStore) handleMsg(message string) (string, error) {
    fmt.Printf("received message %s\n", message)
    m := strings.TrimSpace(message)
    s := strings.Split(m, "|")

    if len(s) < 3 {
        return "FAIL", fmt.Errorf("Incorrect format for string %v", message)
    }

    msg := Message{
        command: s[0],
        pkg:     s[1],
        comment: s[2],
    }
    response := "OK"

    switch msg.command {
    case index:
        ds.pkgInfo[msg.pkg] = msg.comment
        if _, ok := ds.pkgRef[msg.pkg]; ok {
            ds.pkgRef[msg.pkg]++
        } else {
            ds.pkgRef[msg.pkg] = 0
        }
    case remove:
        delete(ds.pkgInfo, msg.pkg)
        if _, ok := ds.pkgRef[msg.pkg]; ok {
            ds.pkgRef[msg.pkg]--
        }
    case count:
        val, ok := ds.pkgRef[msg.pkg]
        if !ok {
            return "FAIL", fmt.Errorf("fail to get count for package %v", msg.pkg)
        }
        response = fmt.Sprintf("%d", val)
    default:
        return "FAIL", fmt.Errorf("go invalid command: %v", msg.command)
    }
    return response, nil
}

and here is a simple program to test concurrent connections:

package main

import (
    "fmt"
    "io/ioutil"
    "math/rand"
    "net"
    "sync"
)

var (
    message = []string{
        "INDEX|vim|comment",
        "REMOVE|vim|blah,blah",
        "INDEX|vim|comment",
        "INDEX|vim|comment",
        "COUNT|vim|comment",
    }
)

func sendMessage(i int) error {
    conn, err := net.Dial("tcp", "localhost:3333")
    if err != nil {
        return fmt.Errorf("error: %v", err)
    }
    defer conn.Close()

    index := rand.Int31n(int32(len(message)))
    _, err = conn.Write([]byte(message[index] + "\n"))
    if err != nil {
        return fmt.Errorf("error: %v", err)
    }

    buf, err := ioutil.ReadAll(conn)
    if err != nil {
        return fmt.Errorf("error: %v", err)
    }
    fmt.Printf("reponse for conn %v: %v", i, string(buf))
    return nil
}

func main() {

    var wg sync.WaitGroup
    nbGoroutines := 3
    wg.Add(nbGoroutines)
    for k := 0; k < nbGoroutines; k++ {
        go func() {
            for i := 1; i <= 100; i++ {
                err := sendMessage(i)
                if err != nil {
                    fmt.Printf("fail: %v\n", err)
                    break
                }
            }
            wg.Done()
        }()
    }
    wg.Wait()
}

---

Other concern

• Using redis could be a good idea if there is really a huge amount of entry in your map, but if you face performance issues, profile your code with pprof to make sure that map accessing is the bottleneck

• if the client disconnect, you won't be able to send the response, so calling Write(...) will just return an error

Hope this helps !