Allow data at queue join (#6182)

* changes

* add changeset

* changes

---------

Co-authored-by: gradio-pr-bot <gradio-pr-bot@users.noreply.github.com>

.changeset/eighty-sides-visit.md

@@ -0,0 +1,5 @@
+---
+"gradio": minor
+---
+
+feat:Allow data at queue join

gradio/queueing.py

@@ -114,7 +114,6 @@ def start(self):
                     ),
                     block_fn.concurrency_limit,
                 )
-        print(">>>", self.concurrency_limit_per_concurrency_id)
 
         run_coro_in_background(self.start_processing)
         run_coro_in_background(self.start_progress_updates)
@@ -452,14 +451,15 @@ async def process_events(self, events: list[Event], batch: bool) -> None:
         awake_events: list[Event] = []
         try:
             for event in events:
-                self.awaiting_data_events[event._id] = event
-                client_awake = await event.get_data()
-                del self.awaiting_data_events[event._id]
-                if client_awake:
-                    event.send_message("process_starts")
-                    awake_events.append(event)
-                else:
-                    await self.clean_event(event)
+                if not event.data:
+                    self.awaiting_data_events[event._id] = event
+                    client_awake = await event.get_data()
+                    del self.awaiting_data_events[event._id]
+                    if not client_awake:
+                        await self.clean_event(event)
+                        continue
+                event.send_message("process_starts")
+                awake_events.append(event)
             if not awake_events:
                 return
             begin_time = time.time()

gradio/routes.py

@@ -579,12 +579,21 @@ async def queue_join(
             session_hash: str,
             request: fastapi.Request,
             username: str = Depends(get_current_user),
+            data: Optional[str] = None,
         ):
             blocks = app.get_blocks()
             if blocks._queue.server_app is None:
                 blocks._queue.set_server_app(app)
 
             event = Event(session_hash, fn_index, request, username)
+            if data is not None:
+                input_data = json.loads(data)
+                event.data = PredictBody(
+                    session_hash=session_hash,
+                    fn_index=fn_index,
+                    data=input_data,
+                    request=request,
+                )
 
             # Continuous events are not put in the queue so that they do not
             # occupy the queue's resource as they are expected to run forever