Merge pull request #32 from giongto35/improve-input-channel

Improve channel

cmd/main.go

@@ -34,8 +34,6 @@ func createOverlordConnection() (*websocket.Conn, error) {
 func initilizeOverlord() {
 	overlord := overlord.NewServer()
 
-	log.Println("http://localhost:8000")
-
 	http.HandleFunc("/", overlord.GetWeb)
 	http.Handle("/static/", http.StripPrefix("/static/", http.FileServer(http.Dir("./static"))))
 
@@ -47,8 +45,6 @@ func initilizeOverlord() {
 	// worker facing port
 	http.HandleFunc("/wso", overlord.WSO)
 	http.ListenAndServe(":8000", nil)
-
-	log.Println("http://localhost:" + *config.Port)
 }
 
 // initializeWorker setup a worker
@@ -65,7 +61,9 @@ func initializeWorker() {
 		log.Println("Close worker")
 		worker.Close()
 	}()
-	worker.Run()
+
+	go worker.Run()
+	http.ListenAndServe(":8001", nil)
 }
 
 func monitor() {

emulator/director.go

@@ -28,7 +28,7 @@ const fps = 60
 func NewDirector(roomID string, imageChannel chan<- *image.RGBA, audioChannel chan<- float32, inputChannel <-chan int) *Director {
 	// TODO: return image channel from where it write
 	director := Director{}
-	director.Done = make(chan struct{})
+	director.Done = make(chan struct{}, 1)
 	director.audioChannel = audioChannel
 	director.imageChannel = imageChannel
 	director.inputChannel = inputChannel

emulator/gameview.go

@@ -85,13 +85,7 @@ func NewGameView(console *nes.Console, title, saveFile string, imageChannel chan
 
 // ListenToInputChannel listen from input channel streamm, which is exposed to WebRTC session
 func (view *GameView) ListenToInputChannel() {
-	for {
-		// Adding ok here make thing slowdown
-		// TODO: Investigate slow
-		keysInBinary, ok := <-view.inputChannel
-		if !ok {
-			return
-		}
+	for keysInBinary := range view.inputChannel {
 		for i := 0; i < NumKeys*2; i++ {
 			b := ((keysInBinary & 1) == 1)
 			view.keyPressed[i] = (view.keyPressed[i] && b) || b

vpx-encoder/encoder.go

@@ -126,15 +126,13 @@ func (v *VpxEncoder) startLooping() {
 		}
 	}()
 
-	for {
-		beginEncoding := time.Now()
-
-		yuv, ok := <-v.Input
-		if !ok || v.Done == true {
+	for yuv := range v.Input {
+		if v.Done == true {
 			// The first time we see IsRunning set to false, we release and return
 			v.Release()
 			return
 		}
+		beginEncoding := time.Now()
 
 		// Add Image
 		v.vpxCodexIter = nil

webrtc/webrtc.go

@@ -59,9 +59,9 @@ func NewWebRTC() *WebRTC {
 	w := &WebRTC{
 		ID: uuid.Must(uuid.NewV4()).String(),
 
-		ImageChannel: make(chan []byte, 2),
+		ImageChannel: make(chan []byte, 100),
 		AudioChannel: make(chan []byte, 1000),
-		InputChannel: make(chan int, 2),
+		InputChannel: make(chan int, 10),
 	}
 	return w
 }
@@ -273,10 +273,9 @@ func (w *WebRTC) startStreaming(vp8Track *webrtc.Track, audioTrack *webrtc.DataC
 			}
 		}()
 
-		for w.isConnected {
-			yuv, ok := <-w.ImageChannel
-			if !ok {
-				log.Println("Screenshot from emulator closed")
+		// TODO: Use same yuv
+		for yuv := range w.ImageChannel {
+			if !w.isConnected {
 				return
 			}
 			if len(w.encoder.Input) < cap(w.encoder.Input) {
@@ -293,13 +292,7 @@ func (w *WebRTC) startStreaming(vp8Track *webrtc.Track, audioTrack *webrtc.DataC
 			}
 		}()
 
-		for w.isConnected {
-			bs, ok := <-w.encoder.Output
-			if !ok {
-				log.Println("WebRTC Video sending Closed")
-				return
-			}
-
+		for bs := range w.encoder.Output {
 			if *config.IsMonitor {
 				log.Println("Encoding FPS : ", w.calculateFPS())
 			}
@@ -315,13 +308,10 @@ func (w *WebRTC) startStreaming(vp8Track *webrtc.Track, audioTrack *webrtc.DataC
 			}
 		}()
 
-		for w.isConnected {
-			data, ok := <-w.AudioChannel
-			if !ok {
-				log.Println("WebRTC Audio sending Closed")
+		for data := range w.AudioChannel {
+			if !w.isConnected {
 				return
 			}
-
 			audioTrack.Send(data)
 		}
 	}()

worker/handlers.go

@@ -109,7 +109,9 @@ func (h *Handler) isRoomRunning(roomID string) bool {
 }
 
 func (h *Handler) Close() {
-	h.oClient.Close()
+	if h.oClient != nil {
+		h.oClient.Close()
+	}
 	// Close all room
 	for _, room := range h.rooms {
 		room.Close()

worker/overlord.go

@@ -165,6 +165,7 @@ func (h *Handler) RouteOverlord() {
 			if ok {
 				session.Close()
 				delete(h.sessions, resp.SessionID)
+				h.detachPeerConn(session.peerconnection)
 			}
 
 			return cws.EmptyPacket

worker/room/media.go

@@ -26,13 +26,7 @@ func (r *Room) startAudio() {
 	var count byte = 0
 
 	// fanout Audio
-	for {
-		sample, ok := <-r.audioChannel
-		if !ok {
-			// Just for guarding
-			log.Println("Warn: Room ", r.ID, " audio channel closed unexpectedly")
-			return
-		}
+	for sample := range r.audioChannel {
 		if !r.IsRunning {
 			log.Println("Room ", r.ID, " audio channel closed")
 			return
@@ -81,13 +75,7 @@ func (r *Room) startVideo() {
 	size := int(float32(config.Width*config.Height) * 1.5)
 	yuv := make([]byte, size, size)
 	// fanout Screen
-	for {
-		image, ok := <-r.imageChannel
-		if !ok {
-			// Just for guarding, should not reached
-			log.Println("Warn: Room ", r.ID, " video channel closed unexpectedly")
-			return
-		}
+	for image := range r.imageChannel {
 		if !r.IsRunning {
 			log.Println("Room ", r.ID, " video channel closed")
 			return

worker/room/room.go

@@ -65,7 +65,7 @@ func NewRoom(roomID, gamepath, gameName string, onlineStorage *storage.Client) *
 		IsRunning:     true,
 		onlineStorage: onlineStorage,
 
-		Done: make(chan struct{}),
+		Done: make(chan struct{}, 1),
 	}
 
 	go room.startVideo()
@@ -121,13 +121,7 @@ func (r *Room) startWebRTCSession(peerconnection *webrtc.WebRTC, playerIndex int
 	}()
 
 	go func() {
-		for {
-			input, ok := <-peerconnection.InputChannel
-			if !ok {
-				return
-				// might consider continue here
-			}
-
+		for input := range peerconnection.InputChannel {
 			if peerconnection.Done || !peerconnection.IsConnected() || !r.IsRunning {
 				return
 			}
@@ -198,6 +192,7 @@ func (r *Room) Close() {
 	close(r.inputChannel)
 	close(r.Done)
 	// Close here is a bit wrong because this read channel
+	// Just dont close it, let it be gc
 	//close(r.imageChannel)
 	//close(r.audioChannel)
 }