diff --git a/emulator/director.go b/emulator/director.go index 7f63cdf4..6eba51a5 100644 --- a/emulator/director.go +++ b/emulator/director.go @@ -24,6 +24,7 @@ type Director struct { 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.audioChannel = audioChannel diff --git a/emulator/gameview.go b/emulator/gameview.go index 67e36918..5a4c270f 100644 --- a/emulator/gameview.go +++ b/emulator/gameview.go @@ -130,6 +130,10 @@ func (view *GameView) Exit() { if cartridge.Battery != 0 { writeSRAM(sramPath(view.saveFile), cartridge.SRAM) } + + // close producer + close(view.imageChannel) + close(view.audioChannel) } // Update is called for every period of time, dt is the elapsed time from the last frame diff --git a/games/Mega Man.nes b/games/Mega Man.nes deleted file mode 100644 index 0669d5f4..00000000 Binary files a/games/Mega Man.nes and /dev/null differ diff --git a/handler/handlers.go b/handler/handlers.go index 0d9d1fe6..c2a536f4 100644 --- a/handler/handlers.go +++ b/handler/handlers.go @@ -107,10 +107,22 @@ func (h *Handler) WS(w http.ResponseWriter, r *http.Request) { }, nil) wssession.BrowserClient.Listen() + + // TODO: Use callback + // If peerconnection is done (client.Done is signalled), we close peerconnection + go func() { + for { + <-client.Done + h.detachPeerConn(wssession.peerconnection) + return + } + }() + } // Detach peerconnection detach/remove a peerconnection from current room func (h *Handler) detachPeerConn(pc *webrtc.WebRTC) { + log.Println("Detach peerconnection") roomID := pc.RoomID room := h.getRoom(roomID) if room == nil { diff --git a/handler/overlord.go b/handler/overlord.go index 76b18ce7..ee353120 100644 --- a/handler/overlord.go +++ b/handler/overlord.go @@ -75,7 +75,7 @@ func (s *Session) RouteOverlord() { "start", func(resp cws.WSPacket) (req cws.WSPacket) { log.Println("Received a start request from overlord") - log.Println("Add the connection to current room on the host") + log.Println("Add the connection to current room on the host ", resp.SessionID) peerconnection := oclient.peerconnections[resp.SessionID] log.Println("start session") diff --git a/handler/room/media.go b/handler/room/media.go index aff2dbb9..4daed757 100644 --- a/handler/room/media.go +++ b/handler/room/media.go @@ -26,54 +26,52 @@ func (r *Room) startAudio() { // fanout Audio for { - select { - case <-r.Done: + sample, ok := <-r.audioChannel + if !ok { + // Just for guarding + log.Println("Warn: Room ", r.ID, " audio channel closed unexpectedly") + return + } + if r.Done { log.Println("Room ", r.ID, " audio channel closed") return - case sample, ok := <-r.audioChannel: - if !ok { - // Just for guarding - log.Println("Warn: Room ", r.ID, " audio channel closed unexpectedly") - return + } + + // TODO: Use worker pool for encoding + pcm[idx] = sample + idx++ + if idx == len(pcm) { + data := make([]byte, 640) + + n, err := enc.EncodeFloat32(pcm, data) + + if err != nil { + log.Println("[!] Failed to decode") + continue } + data = data[:n] + data = append(data, count) - // TODO: Use worker pool for encoding - pcm[idx] = sample - idx++ - if idx == len(pcm) { - data := make([]byte, 640) + // TODO: r.rtcSessions is rarely updated. Lock will hold down perf + //r.sessionsLock.Lock() + for _, webRTC := range r.rtcSessions { + // Client stopped + //if !webRTC.IsClosed() { + //continue + //} - n, err := enc.EncodeFloat32(pcm, data) - - if err != nil { - log.Println("[!] Failed to decode") - continue + // encode frame + // fanout audioChannel + if webRTC.IsConnected() { + // NOTE: can block here + webRTC.AudioChannel <- data } - data = data[:n] - data = append(data, count) - - // TODO: r.rtcSessions is rarely updated. Lock will hold down perf - //r.sessionsLock.Lock() - for _, webRTC := range r.rtcSessions { - // Client stopped - //if !webRTC.IsClosed() { - //continue - //} - - // encode frame - // fanout audioChannel - if webRTC.IsConnected() { - // NOTE: can block here - webRTC.AudioChannel <- data - } - //isRoomRunning = true - } - //r.sessionsLock.Unlock() - - idx = 0 - count = (count + 1) & 0xff - + //isRoomRunning = true } + //r.sessionsLock.Unlock() + + idx = 0 + count = (count + 1) & 0xff } } } @@ -81,35 +79,34 @@ func (r *Room) startAudio() { func (r *Room) startVideo() { // fanout Screen for { - select { - case <-r.Done: + image, ok := <-r.imageChannel + if !ok { + // Just for guarding, should not reached + log.Println("Warn: Room ", r.ID, " video channel closed unexpectedly") + return + } + if r.Done { log.Println("Room ", r.ID, " video channel closed") return - case image, ok := <-r.imageChannel: - if !ok { - // Just for guarding, should not reached - log.Println("Warn: Room ", r.ID, " video channel closed unexpectedly") - return - } - - // TODO: Use worker pool for encoding - yuv := util.RgbaToYuv(image) - // TODO: r.rtcSessions is rarely updated. Lock will hold down perf - //r.sessionsLock.Lock() - for _, webRTC := range r.rtcSessions { - // Client stopped - //if webRTC.IsClosed() { - //continue - //} - - // encode frame - // fanout imageChannel - if webRTC.IsConnected() { - // NOTE: can block here - webRTC.ImageChannel <- yuv - } - } - //r.sessionsLock.Unlock() } + + // TODO: Use worker pool for encoding + yuv := util.RgbaToYuv(image) + // TODO: r.rtcSessions is rarely updated. Lock will hold down perf + //r.sessionsLock.Lock() + for _, webRTC := range r.rtcSessions { + // Client stopped + //if webRTC.IsClosed() { + //continue + //} + + // encode frame + // fanout imageChannel + if webRTC.IsConnected() { + // NOTE: can block here + webRTC.ImageChannel <- yuv + } + } + //r.sessionsLock.Unlock() } } diff --git a/handler/room/room.go b/handler/room/room.go index 20eda172..deddca6f 100644 --- a/handler/room/room.go +++ b/handler/room/room.go @@ -24,7 +24,7 @@ type Room struct { audioChannel chan float32 inputChannel chan int // Done channel is to fire exit event when there is no webRTC session running - Done chan struct{} + Done bool rtcSessions []*webrtc.WebRTC sessionsLock *sync.Mutex @@ -58,7 +58,7 @@ func NewRoom(roomID, gamepath, gameName string, onlineStorage *storage.Client) * rtcSessions: []*webrtc.WebRTC{}, sessionsLock: &sync.Mutex{}, director: director, - Done: make(chan struct{}), + Done: false, onlineStorage: onlineStorage, } @@ -114,34 +114,32 @@ func (r *Room) startWebRTCSession(peerconnection *webrtc.WebRTC, playerIndex int } }() - for { - select { - case <-r.Done: - log.Println("Detach peerconnection from room", r.ID) - return - case <-peerconnection.Done: - r.removeSession(peerconnection) - case input, ok := <-peerconnection.InputChannel: + go func() { + for { + input, ok := <-peerconnection.InputChannel if !ok { return // might consider continue here } + if peerconnection.Done || !peerconnection.IsConnected() || r.Done { + return + } + if peerconnection.IsConnected() { // the first 8 bits belong to player 1 // the next 8 belongs to player 2 ... // We standardize and put it to inputChannel (16 bits) input = input << ((uint(playerIndex) - 1) * emulator.NumKeys) - r.inputChannel <- input - } - default: - if !peerconnection.IsConnected() { - log.Println("peerconnection is closed", peerconnection) - return + select { + case r.inputChannel <- input: + default: + } } } - // Client stopped - } + }() + + log.Println("Peerconn done") } func (r *Room) CleanSession(peerconnection *webrtc.WebRTC) { @@ -150,9 +148,8 @@ func (r *Room) CleanSession(peerconnection *webrtc.WebRTC) { func (r *Room) removeSession(w *webrtc.WebRTC) { fmt.Println("Cleaning session: ", w) - r.sessionsLock.Lock() - defer r.sessionsLock.Unlock() fmt.Println("Sessions list", r.rtcSessions) + // TODO: get list of r.rtcSessions in lock for i, s := range r.rtcSessions { fmt.Println("found session: ", s, w) if s.ID == w.ID { @@ -173,12 +170,16 @@ func (r *Room) removeSession(w *webrtc.WebRTC) { } func (r *Room) Close() { + if r.Done { + return + } + log.Println("Closing room", r.ID) - close(r.Done) log.Println("Closing director of room ", r.ID) close(r.director.Done) log.Println("Closing input of room ", r.ID) close(r.inputChannel) + r.Done = true // Close here is a bit wrong because this read channel //close(r.imageChannel) //close(r.audioChannel) diff --git a/trace?seconds=30 b/trace?seconds=30 new file mode 100644 index 00000000..f4f81220 Binary files /dev/null and b/trace?seconds=30 differ diff --git a/trace?seconds=30.1 b/trace?seconds=30.1 new file mode 100644 index 00000000..ff28a9cf Binary files /dev/null and b/trace?seconds=30.1 differ diff --git a/trace?seconds=30.2 b/trace?seconds=30.2 new file mode 100644 index 00000000..1a5b7560 Binary files /dev/null and b/trace?seconds=30.2 differ diff --git a/webrtc/webrtc.go b/webrtc/webrtc.go index 1b98d8e2..d29bdab1 100644 --- a/webrtc/webrtc.go +++ b/webrtc/webrtc.go @@ -84,7 +84,7 @@ type WebRTC struct { AudioChannel chan []byte InputChannel chan int - Done chan struct{} + Done bool lastTime time.Time curFPS int