Refactor encoder (#75)

* Refactor encoding code

* Close input encoder

config/config.go

@@ -23,8 +23,10 @@ var WSWait = 20 * time.Second
 var MatchWorkerRandom = false
 var ProdEnv = "prod"
 
-//var Codec = CODEC_H264
-var Codec = CODEC_VP8
+var Codec = CODEC_H264
+
+//var Codec = CODEC_VP8
+
 var FileTypeToEmulator = map[string]string{
 	"gba": "gba",
 	"cue": "pcsx",

h264encoder/encoder.go

@@ -77,6 +77,12 @@ func (v *H264Encoder) startLooping() {
 		if r := recover(); r != nil {
 			log.Println("Warn: Recovered panic in encoding ", r)
 		}
+
+		if v.Done == true {
+			// The first time we see IsRunning set to false, we release and return
+			v.release()
+			return
+		}
 	}()
 
 	for img := range v.Input {
@@ -90,12 +96,6 @@ func (v *H264Encoder) startLooping() {
 		v.Output <- v.buf.Bytes()
 		v.buf.Reset()
 	}
-
-	if v.Done == true {
-		// The first time we see IsRunning set to false, we release and return
-		v.release()
-		return
-	}
 }
 
 // Release release memory and stop loop
@@ -105,9 +105,6 @@ func (v *H264Encoder) release() {
 		log.Println("Releasing encoder")
 		// TODO: Bug here, after close it will signal
 		close(v.Output)
-		if v.Input != nil {
-			close(v.Input)
-		}
 		err := v.enc.Close()
 		if err != nil {
 			log.Println("Failed to close H264 encoder")
@@ -129,4 +126,5 @@ func (v *H264Encoder) GetOutputChan() chan []byte {
 // GetDoneChan returns done channel
 func (v *H264Encoder) Stop() {
 	v.Done = true
+	close(v.Input)
 }

libretro/nanoarch/naemulator.go

@@ -122,10 +122,11 @@ func (na *naEmulator) LoadMeta(path string) emulator.Meta {
 
 func (na *naEmulator) Start() {
 	na.playGame(na.gamePath)
-
 	ticker := time.NewTicker(time.Second / 60)
+
 	for range ticker.C {
 		select {
+		// Slow response here
 		case <-na.done:
 			nanoarchShutdown()
 			log.Println("Closed Director")

libretro/nanoarch/nanoarch.go

@@ -61,7 +61,6 @@ var mu sync.Mutex
 var video struct {
 	program uint32
 	vao     uint32
-	texID   uint32
 	pitch   uint32
 	pixFmt  uint32
 	pixType uint32
@@ -120,10 +119,6 @@ func videoConfigure(geom *C.struct_retro_game_geometry) (int, int) {
 
 	fmt.Println("media config", nwidth, nheight, geom.base_width, geom.base_height, geom.aspect_ratio, video.bpp, scale)
 
-	if video.texID == 0 {
-		fmt.Println("Failed to create the video texture")
-	}
-
 	return int(math.Round(nwidth)), int(math.Round(nheight))
 }
 
@@ -239,7 +234,10 @@ func audioWrite2(buf unsafe.Pointer, frames C.size_t) C.size_t {
 
 	for i := 0; i < numFrames; i += 1 {
 		s := float32(pcm[i])
-		NAEmulator.audioChannel <- s
+		select {
+		case NAEmulator.audioChannel <- s:
+		default:
+		}
 	}
 
 	return 2 * frames
@@ -485,10 +483,6 @@ func nanoarchRun() {
 }
 
 func videoSetPixelFormat(format uint32) C.bool {
-	if video.texID != 0 {
-		log.Fatal("Tried to change pixel format after initialization.")
-	}
-
 	switch format {
 	case C.RETRO_PIXEL_FORMAT_0RGB1555:
 		video.pixFmt = gl.UNSIGNED_SHORT_5_5_5_1

vpx-encoder/encoder.go

@@ -126,6 +126,12 @@ func (v *VpxEncoder) startLooping() {
 		if r := recover(); r != nil {
 			log.Println("Warn: Recovered panic in encoding ", r)
 		}
+
+		if v.Done == true {
+			// The first time we see IsRunning set to false, we release and return
+			v.release()
+			return
+		}
 	}()
 
 	size := int(float32(v.width*v.height) * 1.5)
@@ -165,11 +171,6 @@ func (v *VpxEncoder) startLooping() {
 		}
 		v.Output <- bs
 	}
-	if v.Done == true {
-		// The first time we see IsRunning set to false, we release and return
-		v.release()
-		return
-	}
 }
 
 // Release release memory and stop loop

webrtc/webrtc.go

@@ -10,7 +10,6 @@ import (
 	"time"
 
 	"github.com/giongto35/cloud-game/config"
-	"github.com/giongto35/cloud-game/encoder"
 	"github.com/gofrs/uuid"
 	"github.com/pion/webrtc/v2"
 	"github.com/pion/webrtc/v2/pkg/media"
@@ -76,8 +75,7 @@ type WebRTC struct {
 	ID string
 
 	connection *webrtc.PeerConnection
-	//encoder     *vpxEncoder.VpxEncoder
-	encoder     encoder.Encoder
+
 	isConnected bool
 	isClosed    bool
 	// for yuvI420 image
@@ -242,17 +240,12 @@ func (w *WebRTC) StopClient() {
 
 	log.Println("===StopClient===")
 	w.isConnected = false
-	if w.encoder != nil {
-		// NOTE: We signal using bool value instead of channel for better performance
-		w.encoder.Stop()
-	}
 	if w.connection != nil {
 		w.connection.Close()
 	}
 	w.connection = nil
-	close(w.InputChannel)
+	//close(w.InputChannel)
 	// webrtc is producer, so we close
-	close(w.encoder.GetInputChan())
 	// NOTE: ImageChannel is waiting for input. Close in writer is not correct for this
 	close(w.ImageChannel)
 	close(w.AudioChannel)
@@ -270,7 +263,7 @@ func (w *WebRTC) startStreaming(vp8Track *webrtc.Track, opusTrack *webrtc.Track)
 	go func() {
 		defer func() {
 			if r := recover(); r != nil {
-				fmt.Println("Recovered when sent to closed encoder output channel")
+				fmt.Println("Recovered from err", r)
 			}
 		}()
 
@@ -289,7 +282,7 @@ func (w *WebRTC) startStreaming(vp8Track *webrtc.Track, opusTrack *webrtc.Track)
 	go func() {
 		defer func() {
 			if r := recover(); r != nil {
-				fmt.Println("Recovered when sent to closed Audio Channel")
+				fmt.Println("Recovered from err", r)
 			}
 		}()
 

worker/handlers.go

@@ -102,6 +102,7 @@ func (h *Handler) detachPeerConn(pc *webrtc.WebRTC) {
 			// Signal end of input Channel
 			log.Println("Signal input chan")
 			pc.InputChannel <- -1
+			close(pc.InputChannel)
 		}
 	}
 }

worker/room/media.go

@@ -108,6 +108,10 @@ func (r *Room) startVideo(width, height int) {
 		encoder, err = vpxencoder.NewVpxEncoder(width, height, 20, 1200, 5)
 	}
 
+	defer func() {
+		encoder.Stop()
+	}()
+
 	if err != nil {
 		fmt.Println("error create new encoder", err)
 		return
@@ -123,27 +127,27 @@ func (r *Room) startVideo(width, height int) {
 			}
 		}()
 
-		for image := range r.imageChannel {
-			if !r.IsRunning {
-				log.Println("Room ", r.ID, " video channel closed")
-				return
-			}
-			if len(einput) < cap(einput) {
-				einput <- image
+		// fanout Screen
+		for data := range eoutput {
+			// TODO: r.rtcSessions is rarely updated. Lock will hold down perf
+			for _, webRTC := range r.rtcSessions {
+				// encode frame
+				// fanout imageChannel
+				if webRTC.IsConnected() {
+					// NOTE: can block here
+					webRTC.ImageChannel <- data
+				}
 			}
 		}
 	}()
 
-	// fanout Screen
-	for data := range eoutput {
-		// TODO: r.rtcSessions is rarely updated. Lock will hold down perf
-		for _, webRTC := range r.rtcSessions {
-			// encode frame
-			// fanout imageChannel
-			if webRTC.IsConnected() {
-				// NOTE: can block here
-				webRTC.ImageChannel <- data
-			}
+	for image := range r.imageChannel {
+		if !r.IsRunning {
+			log.Println("Room ", r.ID, " video channel closed")
+			return
+		}
+		if len(einput) < cap(einput) {
+			einput <- image
 		}
 	}
 }

worker/room/room.go

@@ -219,7 +219,6 @@ func (r *Room) Close() {
 	log.Println("Closing room", r.ID)
 	log.Println("Closing director of room ", r.ID)
 	r.director.Close()
-	//close(r.director.Done)
 	log.Println("Closing input of room ", r.ID)
 	close(r.inputChannel)
 	close(r.Done)