Merge pull request #25 from giongto35/channel-free

Channel free

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

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

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 {

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")

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()
 	}
 }

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)

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