socket-io: Allocated memory isn't released under load test

[tavancini-tc]

Hi Nika! So, I'm running some load tests here and noticed the server is not releasing allocated memory.

Here is what I found:

1. After running the load test, I got around 5Gib of allocated memory that is not released even after GC execution.

2. Digging into pprof, I got:

To reproduce the test:

1. Install k6;
2. Start the server go run main.go (code below);
3. Run k6 load test script k6 run loadTest.js (code below);
4. Run pprof:

   $ go tool pprof http://localhost:8003/debug/pprof/heap

main.go

package main

import (
    "context"
    "fmt"
    "log"
    "net/http"
    "net/http/pprof"
    "os"
    "os/signal"
    "runtime"
    "strings"
    "time"

    sio "github.com/njones/socketio"
    "github.com/njones/socketio/callback"
    eio "github.com/njones/socketio/engineio"
    ser "github.com/njones/socketio/serialize"
)

const (
    httpServerPort = ":8003"
)

func main() {
    var ruTimerQuitChan chan struct{}

    sioServer := sio.NewServerV2(
        sio.WithPath("/websocket/api/socket.io/"),
        eio.WithCors(eio.CORSorigin{"*"}),
        eio.WithSessionShave(1*time.Millisecond),
        eio.WithPingInterval(5*time.Second),
        eio.WithPingTimeout(1*time.Minute),
        eio.WithMaxPayload(1000000),
    )

    // /
    sioServer.OnConnect(func(socket *sio.SocketV2) error {
        log.Printf("/ connected: %s", socket.ID().String())
        log.Printf("/ token: %s", socket.Request().URL.Query().Get("token"))
        return nil
    })

    // CHANNEL
    channel := sioServer.Of("/channel")
    channel.OnConnect(func(socket *sio.SocketV2) error {
        log.Printf("/channel connected: %s", socket.ID().String())
        log.Printf("/channel token: %s", socket.Request().URL.Query().Get("token"))
        return nil
    })

    // CHAT
    chat := sioServer.Of("/chat")
    chat.OnConnect(func(socket *sio.SocketV2) error {
        log.Printf("/chat connected: %s", socket.ID().String())
        log.Printf("/chat token: %s", socket.Request().URL.Query().Get("token"))
        socket.On("join", callback.Wrap{
            Parameters: []ser.Serializable{ser.StrParam},
            Func: func() interface{} {
                return func(room string) error {
                    log.Print("/chat join event")
                    return nil
                }
            },
        })

        socket.On("leave", callback.Wrap{
            Parameters: []ser.Serializable{ser.StrParam},
            Func: func() interface{} {
                return func(room string) error {
                    log.Print("/chat leave event")
                    return nil
                }
            },
        })

        return nil
    })

    // Debug
    ruTicker := time.NewTicker(time.Second * 10)
    ruTimerQuitChan = make(chan struct{})
    go func() {
        for {
            select {
            case <-ruTicker.C:
                printMemStats()
            case <-ruTimerQuitChan:
                ruTicker.Stop()
                return
            }
        }
    }()

    defaultMux := http.NewServeMux()

    // Socket.io setup
    defaultMux.Handle("/", sioServer)

    // Pprof setup
    defaultMux.HandleFunc("/debug/pprof/", pprof.Index)

    sioHTTPServer := &http.Server{
        Addr:              httpServerPort,
        Handler:           defaultMux,
        ReadHeaderTimeout: 2 * time.Second,
    }

    // Start server
    log.Printf("Server running at %s.", httpServerPort)
    go func() {
        if err := sioHTTPServer.ListenAndServe(); err != nil {
            log.Fatal(err)
        }
    }()

    stop := make(chan os.Signal, 1)
    signal.Notify(stop, os.Interrupt)
    <-stop
    close(ruTimerQuitChan)
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()
    if err := sioHTTPServer.Shutdown(ctx); err != nil {
        log.Fatal(err)
    }
}

func printMemStats() {
    var stats strings.Builder
    var m runtime.MemStats
    runtime.ReadMemStats(&m)
    stats.WriteString(fmt.Sprintf("Alloc=%vMiB; TotalAlloc=%v MiB; Sys=%vMiB; GoRoutines=%d; NumGC=%v",
        bToMb(m.Alloc),
        bToMb(m.TotalAlloc),
        bToMb(m.Sys),
        runtime.NumGoroutine(),
        m.NumGC))
    log.Print(stats.String())
}

func bToMb(b uint64) uint64 {
    return b / 1024 / 1024
}

loadTest.js

import { check, group, sleep } from 'k6';
import ws from 'k6/ws';

export const options = {
  thresholds: {
    http_req_failed: ['rate<0.50'],
    http_req_duration: ['p(90)<5000'],
  },
  stages: [
    { duration: '2m', target: 100 },
    { duration: '4m', target: 100 },
    { duration: '2m', target: 0 },
  ],
};

const nameSpaces = ['', 'channel', 'chat'];
const token = '01234567890123456789012345';
const url = 'ws://localhost:8003/websocket/api/socket.io/?token=' + token + '&EIO=3&transport=websocket';

export default function () {
  nameSpaces.forEach((nameSpace) => {
    group("Test '" + nameSpace + "'", function () {
      let response = ws.connect(url, {}, function (socket) {
        socket.on('open', function open() {
          if (nameSpace.length > 0) {
            socket.send('40/' + nameSpace);
          }

          if (nameSpace === 'chat') {
            sleep(1);
            socket.send('42/chat,["join","room-id"]');
            sleep(1);
            socket.send('42/chat,["leave","room-id"]');
          }

          sleep(1);
          socket.close();
        });
      });

      check(response, {
        'status is 101': (r) => r && r.status === 101,
      });
    });
  });
}

[njones]

Interesting... I'll check this out, and fix it.

Thanks again for the awesome detailed report!

[tavancini-tc]

Hi Nika, how are you? Do you have any updates? Tks