Add blocking mechanism: only one player can use stream at a time

app.py

@@ -66,6 +66,7 @@ server_ready = False    # <--- readiness flag
 stream_lock = threading.Lock()
 stream_thread = None
 stream_running = False
+current_player_sid = None  # Track which session is currently playing
 latest_action = 1  # 0=init, 1=nothing, 2=up, 3=down
 target_fps = 30
 frame_index = 0
@@ -368,13 +369,16 @@ def generate_frames():
 # --------------------------
 # Socket events & helpers
 # --------------------------
-def start_stream(n_steps=8, cfg=0.0, fps=30, clamp=True):
-    global stream_thread, stream_running, frame_index, target_fps, latest_action
+def start_stream(n_steps=8, cfg=0.0, fps=30, clamp=True, player_sid=None):
+    global stream_thread, stream_running, frame_index, target_fps, latest_action, current_player_sid
     if not server_ready:
         _broadcast_ready()
         raise RuntimeError("Server not ready")
     with stream_lock:
+        if stream_running and current_player_sid is not None:
+            raise RuntimeError(f"Another player is already using the stream (session: {current_player_sid[:8]}...)")
         stop_stream()
+        current_player_sid = player_sid
         target_fps = int(fps)
         frame_index = 0
         _reset_cache_fresh()
@@ -384,13 +388,16 @@ def start_stream(n_steps=8, cfg=0.0, fps=30, clamp=True):
         stream_thread.start()
 
 def stop_stream():
-    global stream_thread, stream_running
+    global stream_thread, stream_running, current_player_sid
     with stream_lock:
         if stream_thread is not None:
             stream_thread.stop()
             stream_thread.join(timeout=1.0)
             stream_thread = None
         stream_running = False
+        current_player_sid = None
+        # Notify all clients that the stream is available
+        socketio.emit('stream_available', {'status': 'available'})
 
 @socketio.on_error_default
 def default_error_handler(e):
@@ -423,24 +430,44 @@ def handle_connect():
 def handle_disconnect(*args):
     sid = request.sid
     print(f'Client disconnected: {sid}')
-    # Note: We don't stop the stream on disconnect since multiple users can be connected
+    # If the current player disconnects, free up the stream
+    with stream_lock:
+        if current_player_sid == sid:
+            print(f'Current player disconnected, freeing stream')
+            stop_stream()
 
 @socketio.on('start_stream')
 def handle_start_stream(data):
     try:
+        sid = request.sid
         if not server_ready:
             # Tell client to keep showing spinner
             emit('server_status', {'ready': server_ready, 'busy': False})
             return
         
+        # Check if someone else is already playing
+        with stream_lock:
+            if stream_running and current_player_sid is not None and current_player_sid != sid:
+                emit('error', {'message': 'Another player is currently using the stream. Please wait for them to finish.'})
+                emit('stream_busy', {'current_player': current_player_sid[:8] if current_player_sid else 'unknown'})
+                return
+        
         n_steps = int(data.get('n_steps', 8))
         cfg = float(data.get('cfg', 0))
         fps = int(data.get('fps', 30))
         clamp = bool(data.get('clamp', True))
-        print(f"Starting stream @ {fps} FPS (n_steps={n_steps}, cfg={cfg}, clamp={clamp})")
+        print(f"Starting stream @ {fps} FPS (n_steps={n_steps}, cfg={cfg}, clamp={clamp}) for session {sid[:8]}")
         try:
-            start_stream(n_steps=n_steps, cfg=cfg, fps=fps, clamp=clamp)
+            start_stream(n_steps=n_steps, cfg=cfg, fps=fps, clamp=clamp, player_sid=sid)
             emit('stream_started', {'status': 'ok'})
+            # Notify other clients that stream is now busy
+            socketio.emit('stream_busy', {'current_player': sid[:8]}, include_self=False)
+        except RuntimeError as e:
+            if "already using" in str(e):
+                emit('error', {'message': str(e)})
+                emit('stream_busy', {'current_player': current_player_sid[:8] if current_player_sid else 'unknown'})
+            else:
+                raise
         except Exception as e:
             print(f"Error starting stream: {e}")
             import traceback
@@ -454,16 +481,27 @@ def handle_start_stream(data):
 
 @socketio.on('action')
 def handle_action(data):
-    global latest_action
-    action = int(data.get('action', 1))
+    global latest_action, current_player_sid
+    sid = request.sid
+    # Only allow actions from the current player
     with stream_lock:
+        if current_player_sid != sid:
+            emit('error', {'message': 'You are not the current player. Please wait your turn.'})
+            return
+        action = int(data.get('action', 1))
         latest_action = action
     emit('action_ack', {'received': action, 'will_apply_to_frame_index': frame_index})
 
 @socketio.on('stop_stream')
 def handle_stop_stream():
-    print('Stopping stream')
+    sid = request.sid
+    print(f'Stopping stream for session {sid[:8]}')
     try:
+        # Only allow the current player to stop
+        with stream_lock:
+            if current_player_sid != sid:
+                emit('error', {'message': 'You are not the current player. Only the current player can stop the stream.'})
+                return
         stop_stream()
         emit('stream_stopped', {'status': 'ok'})
     except Exception as e:

static/index.html

@@ -56,6 +56,9 @@
       <div id="fpsDisplay" style="margin-top: 8px; font-size: 16px; font-family: monospace;">
         Achieved FPS: <span id="fpsValue">-</span>
       </div>
+      <div id="waitingMessage" style="margin-top: 12px; padding: 8px; background: #333; border-radius: 4px; display: none; color: #ffa500;">
+        ⏳ Another player is currently using the stream. Please wait for them to finish.
+      </div>
       <div>
         This is the output of a small frame-autoregressive transformer trained with rectified flow matching to simulate pong frames conditioned on user inputs for the blue paddle. It should reach 12 FPS when using 4 steps for generation unless something else is running on my machine.
       </div>
@@ -114,6 +117,20 @@
 
       const actionValue = document.getElementById('actionValue');
       const fpsValue = document.getElementById('fpsValue');
+      const waitingMessage = document.getElementById('waitingMessage');
+
+      // Handle stream busy/available events
+      socket.on('stream_busy', (data) => {
+        console.log('Stream is busy:', data);
+        waitingMessage.style.display = 'block';
+        startBtn.disabled = true;
+      });
+
+      socket.on('stream_available', (data) => {
+        console.log('Stream is available:', data);
+        waitingMessage.style.display = 'none';
+        startBtn.disabled = false;
+      });
 
       // Incoming frames
       socket.on('frame', ({ frame, frame_index, action, fps }) => {
@@ -129,10 +146,16 @@
 
       socket.on('error', (e) => {
         console.warn('server error', e);
-        // The server_status event will handle showing the appropriate overlay
-        // Just log the error for now
         if (e && e.message) {
           console.error('Server error message:', e.message);
+          // Show error message to user
+          if (e.message.includes('Another player') || e.message.includes('not the current player')) {
+            waitingMessage.style.display = 'block';
+            waitingMessage.textContent = '⏳ ' + e.message;
+            startBtn.disabled = true;
+          } else {
+            alert('Error: ' + e.message);
+          }
         }
       });