From bfc40f694040ab5175fc683e3e2e0518617d3996 Mon Sep 17 00:00:00 2001 From: giongto35 Date: Mon, 20 May 2019 15:16:56 +0800 Subject: [PATCH 1/2] Change done chan to 1 --- cmd/main.go | 8 +++----- emulator/director.go | 2 +- webrtc/webrtc.go | 5 +++-- worker/handlers.go | 4 +++- worker/room/room.go | 2 +- 5 files changed, 11 insertions(+), 10 deletions(-) diff --git a/cmd/main.go b/cmd/main.go index 17e6b94d..b34ee26e 100644 --- a/cmd/main.go +++ b/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() { diff --git a/emulator/director.go b/emulator/director.go index 9c911e54..70ed0535 100644 --- a/emulator/director.go +++ b/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 diff --git a/webrtc/webrtc.go b/webrtc/webrtc.go index bb555781..d6dabc93 100644 --- a/webrtc/webrtc.go +++ b/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,6 +273,7 @@ func (w *WebRTC) startStreaming(vp8Track *webrtc.Track, audioTrack *webrtc.DataC } }() + // TODO: Use same yuv for w.isConnected { yuv, ok := <-w.ImageChannel if !ok { diff --git a/worker/handlers.go b/worker/handlers.go index 7725b81d..423a437f 100644 --- a/worker/handlers.go +++ b/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() diff --git a/worker/room/room.go b/worker/room/room.go index 8e855a17..0aa40a32 100644 --- a/worker/room/room.go +++ b/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() From b8cffcbc6715be979b9cfd9f2b60130c58460c1a Mon Sep 17 00:00:00 2001 From: giongto35 Date: Mon, 20 May 2019 16:00:09 +0800 Subject: [PATCH 2/2] use for to wait channel --- emulator/gameview.go | 8 +------- vpx-encoder/encoder.go | 8 +++----- webrtc/webrtc.go | 21 +++++---------------- worker/overlord.go | 1 + worker/room/media.go | 16 ++-------------- worker/room/room.go | 9 ++------- 6 files changed, 14 insertions(+), 49 deletions(-) diff --git a/emulator/gameview.go b/emulator/gameview.go index 0be6c22b..8c7113e2 100644 --- a/emulator/gameview.go +++ b/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 diff --git a/vpx-encoder/encoder.go b/vpx-encoder/encoder.go index d295bceb..3f85f816 100644 --- a/vpx-encoder/encoder.go +++ 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 diff --git a/webrtc/webrtc.go b/webrtc/webrtc.go index d6dabc93..4d416673 100644 --- a/webrtc/webrtc.go +++ b/webrtc/webrtc.go @@ -274,10 +274,8 @@ func (w *WebRTC) startStreaming(vp8Track *webrtc.Track, audioTrack *webrtc.DataC }() // TODO: Use same yuv - for w.isConnected { - yuv, ok := <-w.ImageChannel - if !ok { - log.Println("Screenshot from emulator closed") + for yuv := range w.ImageChannel { + if !w.isConnected { return } if len(w.encoder.Input) < cap(w.encoder.Input) { @@ -294,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()) } @@ -316,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) } }() diff --git a/worker/overlord.go b/worker/overlord.go index 00c5e8d6..13c0253f 100644 --- a/worker/overlord.go +++ b/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 diff --git a/worker/room/media.go b/worker/room/media.go index 24fe7083..27a354df 100644 --- a/worker/room/media.go +++ b/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 diff --git a/worker/room/room.go b/worker/room/room.go index 0aa40a32..1e45e0bb 100644 --- a/worker/room/room.go +++ b/worker/room/room.go @@ -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) }