Speed up rendering and encoding

Author: vkhanhqui

013-cloud-gaming/game/snake_renderer.go

@@ -4,6 +4,7 @@ import (
 	"image"
 	"image/color"
 	"log"
+	"sync"
 	"time"
 )
 
@@ -13,7 +14,10 @@ var (
 	foodColor      = color.RGBA{R: 0, G: 255, B: 0, A: 255}
 )
 
-const bytesPerPixel = 3 // RGB: 3 bytes per pixel
+const (
+	numWorkers    = 4
+	bytesPerPixel = 3 // RGB: 3 bytes per pixel
+)
 
 func StartFrameRenderer(gameStateCh chan *Snake, canvasCh chan *Canvas) {
 	defer func() {
@@ -22,33 +26,44 @@ func StartFrameRenderer(gameStateCh chan *Snake, canvasCh chan *Canvas) {
 		}
 	}()
 
-	for {
-		start := time.Now()
-		gameState, ok := <-gameStateCh
-		if !ok {
-			break
-		}
+	var wg sync.WaitGroup
+	workerCh := make(chan *Snake)
+
+	for i := 0; i < numWorkers; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for {
+				gameState, ok := <-workerCh
+				if !ok {
+					return
+				}
 
-		img := image.NewRGBA(image.Rect(0, 0, FRAME_WIDTH, FRAME_HEIGHT))
-		matrix := gameState.GetMatrix()
+				img := image.NewRGBA(image.Rect(0, 0, FRAME_WIDTH, FRAME_HEIGHT))
+				matrix := gameState.GetMatrix()
 
-		for y := 0; y < len(matrix); y++ {
-			for x := 0; x < len(matrix[0]); x++ {
-				rectMin := image.Point{X: x * CHUNK_SIZE, Y: y * CHUNK_SIZE}
-				rectMax := image.Point{X: rectMin.X + CHUNK_SIZE, Y: rectMin.Y + CHUNK_SIZE}
-				switch matrix[y][x] {
-				case HEAD:
-					drawRectangle(img, rectMin, rectMax, snakeHeadColor)
-				case BODY:
-					drawRectangle(img, rectMin, rectMax, snakeBodyColor)
-				case FOOD:
-					drawRectangle(img, rectMin, rectMax, foodColor)
+				for y := 0; y < len(matrix); y++ {
+					for x := 0; x < len(matrix[0]); x++ {
+						rectMin := image.Point{X: x * CHUNK_SIZE, Y: y * CHUNK_SIZE}
+						rectMax := image.Point{X: rectMin.X + CHUNK_SIZE, Y: rectMin.Y + CHUNK_SIZE}
+						switch matrix[y][x] {
+						case HEAD:
+							drawRectangle(img, rectMin, rectMax, snakeHeadColor)
+						case BODY:
+							drawRectangle(img, rectMin, rectMax, snakeBodyColor)
+						case FOOD:
+							drawRectangle(img, rectMin, rectMax, foodColor)
+						}
+					}
 				}
+
+				canvasCh <- &Canvas{Data: convertRGBAtoRGB(img), Timestamp: time.Now()}
 			}
-		}
+		}()
+	}
 
-		canvasCh <- &Canvas{Data: convertRGBAtoRGB(img), Timestamp: time.Now()}
-		log.Println("StartFrameRenderer", time.Since(start))
+	for gameState := range gameStateCh {
+		workerCh <- gameState
 	}
 }
 

013-cloud-gaming/service/encoder_v2.go

@@ -0,0 +1,137 @@
+package service
+
+import (
+	"cloud-gaming/game"
+	"fmt"
+	"io"
+	"log"
+	"os"
+	"os/exec"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+type Encoder struct {
+	encodedFrameCh chan *Streamable
+	canvasCh       <-chan *game.Canvas
+	closeSignal    <-chan bool
+	windowWidth    int
+	windowHeight   int
+	closed         int32
+	wg             sync.WaitGroup
+}
+
+type EncoderOptions struct {
+	EncodedFrameChannel chan *Streamable
+	CanvasChannel       <-chan *game.Canvas
+	CloseSignal         <-chan bool
+	WindowWidth         int
+	WindowHeight        int
+}
+
+type Streamable struct {
+	Data      []byte
+	Timestamp time.Time
+}
+
+func NewEncoder(options *EncoderOptions) *Encoder {
+	return &Encoder{
+		encodedFrameCh: options.EncodedFrameChannel,
+		canvasCh:       options.CanvasChannel,
+		closeSignal:    options.CloseSignal,
+		windowWidth:    options.WindowWidth,
+		windowHeight:   options.WindowHeight,
+	}
+}
+
+func (e *Encoder) isClosed() bool {
+	return atomic.LoadInt32(&e.closed) == 1
+}
+
+func (e *Encoder) markAsClosed() {
+	atomic.StoreInt32(&e.closed, 1)
+}
+
+const ffmpegBaseCommand = "ffmpeg -hide_banner -loglevel error -threads 0 -re -f rawvideo -pixel_format rgb24 -video_size %dx%d -framerate %v -r %v -i pipe:0 -pix_fmt yuv420p -c:v h264_videotoolbox -f h264 pipe:1"
+
+func (e *Encoder) Start() {
+	ffmpegCommand := fmt.Sprintf(ffmpegBaseCommand, e.windowWidth, e.windowHeight, game.FPS, game.FPS)
+	cmd := exec.Command("bash", "-c", ffmpegCommand)
+	cmd.Stderr = os.Stderr
+	inPipe, err := cmd.StdinPipe()
+	if err != nil {
+		panic(err)
+	}
+
+	outPipe, err := cmd.StdoutPipe()
+	if err != nil {
+		panic(err)
+	}
+
+	err = cmd.Start()
+	if err != nil {
+		panic(err)
+	}
+
+	e.wg.Add(2)
+
+	go e.writeToFFmpeg(inPipe)
+	go e.streamToWebRTCTrack(outPipe)
+
+	go func() {
+		_, ok := <-e.closeSignal
+		if !ok {
+			e.markAsClosed()
+			fmt.Println("Closing encoder")
+
+			cmd.Wait()
+			e.wg.Wait()
+			close(e.encodedFrameCh)
+		}
+	}()
+
+}
+
+func (e *Encoder) writeToFFmpeg(inPipe io.WriteCloser) {
+	defer e.wg.Done()
+
+	for canvas := range e.canvasCh {
+		if e.isClosed() {
+			return
+		}
+		select {
+		case <-e.encodedFrameCh: // Check if there's a backlog.
+			continue
+		default:
+			_, err := inPipe.Write(canvas.Data)
+			if err != nil {
+				panic(err)
+			}
+		}
+	}
+
+	inPipe.Close()
+}
+
+func (e *Encoder) streamToWebRTCTrack(outPipe io.Reader) {
+	defer e.wg.Done()
+
+	buf := make([]byte, 1024*8)
+	for {
+		if e.isClosed() {
+			return
+		}
+
+		timestamp := time.Now()
+		n, err := outPipe.Read(buf)
+		if err == io.EOF {
+			break
+		} else if err != nil {
+			log.Println(err)
+			continue
+		}
+
+		e.encodedFrameCh <- &Streamable{Data: buf[:n], Timestamp: timestamp}
+	}
+}

013-cloud-gaming/service/worker.go

@@ -41,7 +41,14 @@ func (w *Worker) onDataChannel(p *PeerConnState) {
 		go gameLoop.Start()
 		go game.StartFrameRenderer(gameStateCh, canvasCh)
 
-		go w.startEncoder(canvasCh, encodedFrameCh)
+		encoder := NewEncoder(&EncoderOptions{
+			EncodedFrameChannel: encodedFrameCh,
+			CanvasChannel:       canvasCh,
+			WindowWidth:         game.FRAME_WIDTH,
+			WindowHeight:        game.FRAME_HEIGHT,
+		})
+		encoder.Start()
+
 		go w.startStreaming(encodedFrameCh, senders)
 		go w.closeConnection(closeSignal, dataCh, p, gameStateCh, cmdCh, canvasCh, encodedFrameCh)