server: faster json encoding

Author: iwanhae

pkg/server/event.go

@@ -1,7 +1,11 @@
 package server
 
 import (
+	"bytes"
+	"context"
 	"fmt"
+	"runtime"
+	"sync"
 
 	"github.com/iwanhae/kuview/pkg/controller"
 	"github.com/labstack/echo/v4"
@@ -43,11 +47,8 @@ func (s *Server) subscribe(c echo.Context) error {
 	}()
 
 	// 2. Send the snapshot to the client.
-	for _, v := range cache {
-		evt := Event{
-			Data: eventAsJSON(v),
-		}
-		if err := evt.MarshalTo(w); err != nil {
+	for v := range s.encodeEventsParallel(c.Request().Context(), cache) {
+		if _, err := w.Write(v); err != nil {
 			return err
 		}
 	}
@@ -92,3 +93,47 @@ func (s *Server) Emit(v *controller.Event) {
 	// it must be sent after the cache is updated
 	s.evtCh <- v
 }
+
+func (s *Server) encodeEventsParallel(ctx context.Context, cache []*controller.Event) <-chan []byte {
+	ch := make(chan []byte, 10*runtime.NumCPU())
+
+	q := make(chan *controller.Event, 10*runtime.NumCPU())
+	go func() {
+		defer close(q)
+		for _, v := range cache {
+			q <- v
+		}
+	}()
+
+	var wg sync.WaitGroup
+	for i := 0; i < runtime.NumCPU(); i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for {
+				select {
+				case <-ctx.Done():
+					return
+				case v, ok := <-q:
+					if !ok {
+						return
+					}
+					evt := Event{Data: eventAsJSON(v)}
+					buf := bytes.NewBuffer(nil)
+
+					if err := evt.MarshalTo(buf); err != nil {
+						log.Error().Err(err).Msg("failed to marshal event to buffer")
+						continue
+					}
+					ch <- buf.Bytes()
+				}
+			}
+		}()
+	}
+	go func() {
+		wg.Wait()
+		close(ch)
+	}()
+
+	return ch
+}