Release new queue beta

.circleci/config.yml

@@ -58,7 +58,7 @@ jobs:
       - run:
           command: |
             . venv/bin/activate
-            python -m  flake8 --ignore=E731,E501,E722,W503,E126,F401,E203,F403 gradio test
+            python -m  flake8 --ignore=E731,E501,E722,W503,E126,F401,E203,F403,F541 gradio test
       - run:
           command: |
             . venv/bin/activate

demo/fake_gan_2/run.py

@@ -8,7 +8,7 @@
 
 
 def fake_gan(*args):
-    time.sleep(1)
+    time.sleep(8)
     image = random.choice(
         [
             "https://images.unsplash.com/photo-1507003211169-0a1dd7228f2d?ixlib=rb-1.2.1&ixid=MnwxMjA3fDB8MHxwaG90by1wYWdlfHx8fGVufDB8fHx8&auto=format&fit=crop&w=387&q=80",
@@ -36,6 +36,7 @@ def fake_gan(*args):
         [os.path.join(os.path.dirname(__file__), "files/zebra.jpg")],
     ],
 )
+demo.queue()
 
 if __name__ == "__main__":
     demo.launch()

demo/queue_concurrench/run.py

@@ -0,0 +1,16 @@
+import gradio as gr
+import asyncio
+
+
+async def say_hello(name):
+  await asyncio.sleep(5)
+  return f"Hello {name}!"
+
+
+with gr.Blocks() as demo:
+  inp = gr.Textbox()
+  outp = gr.Textbox()
+  button = gr.Button()
+  button.click(say_hello, inp, outp)
+
+  demo.configure_queue(concurrency_count=5).launch(enable_queue=True)

gradio/blocks.py

@@ -7,19 +7,21 @@
 import random
 import sys
 import time
+import warnings
 import webbrowser
 from types import ModuleType
 from typing import TYPE_CHECKING, Any, AnyStr, Callable, Dict, List, Optional, Tuple
 
 import anyio
+import requests
 from anyio import CapacityLimiter
 
 from gradio import (
     components,
     encryptor,
+    event_queue,
     external,
     networking,
-    queueing,
     routes,
     strings,
     utils,
@@ -329,6 +331,7 @@ def __init__(
         self.requires_permissions = False  # TODO: needs to be implemented
         self.encrypt = False
         self.share = False
+        self.enable_queue = None
         if css is not None and os.path.exists(css):
             with open(css) as css_file:
                 self.css = css_file.read()
@@ -664,6 +667,7 @@ def get_config_file(self):
             "theme": self.theme,
             "css": self.css,
             "title": self.title or "Gradio",
+            "is_space": self.is_space,
             "enable_queue": getattr(
                 self, "enable_queue", False
             ),  # attribute set at launch
@@ -765,6 +769,35 @@ def clear(self):
         self.children = []
         return self
 
+    @document()
+    def queue(
+        self,
+        concurrency_count: int = 1,
+        status_update_rate: float | str = "auto",
+        client_position_to_load_data: int = 30,
+        default_enabled: bool = True,
+    ):
+        """
+        You can control the rate of processed requests by creating a queue. This will allow you to set the number of requests to be processed at one time, and will let users know their position in the queue.
+        Parameters:
+            concurrency_count: Number of worker threads that will be processing requests concurrently.
+            status_update_rate: If "auto", Queue will send status estimations to all clients whenever a job is finished. Otherwise Queue will send status at regular intervals set by this parameter as the number of seconds.
+            client_position_to_load_data: Once a client's position in Queue is less that this value, the Queue will collect the input data from the client. You may make this smaller if clients can send large volumes of data, such as video, since the queued data is stored in memory.
+            default_enabled: If True, all event listeners will use queueing by default.
+        Example:
+            demo = gr.Interface(gr.Textbox(), gr.Image(), image_generator)
+            demo.queue(concurrency_count=3)
+            demo.launch()
+        """
+        self.enable_queue = default_enabled
+        event_queue.Queue.configure_queue(
+            live_updates=status_update_rate == "auto",
+            concurrency_count=concurrency_count,
+            data_gathering_start=client_position_to_load_data,
+            update_intervals=status_update_rate if status_update_rate != "auto" else 1,
+        )
+        return self
+
     def launch(
         self,
         inline: bool = None,
@@ -806,7 +839,7 @@ def launch(
             server_port: will start gradio app on this port (if available). Can be set by environment variable GRADIO_SERVER_PORT. If None, will search for an available port starting at 7860.
             server_name: to make app accessible on local network, set this to "0.0.0.0". Can be set by environment variable GRADIO_SERVER_NAME. If None, will use "127.0.0.1".
             show_tips: if True, will occasionally show tips about new Gradio features
-            enable_queue: if True, inference requests will be served through a queue instead of with parallel threads. Required for longer inference times (> 1min) to prevent timeout. The default option in HuggingFace Spaces is True. The default option elsewhere is False.
+            enable_queue: DEPRECATED (use .queue() method instead.) if True, inference requests will be served through a queue instead of with parallel threads. Required for longer inference times (> 1min) to prevent timeout. The default option in HuggingFace Spaces is True. The default option elsewhere is False.
             max_threads: allow up to `max_threads` to be processed in parallel. The default is inherited from the starlette library (currently 40).
             width: The width in pixels of the iframe element containing the interface (used if inline=True)
             height: The height in pixels of the iframe element containing the interface (used if inline=True)
@@ -842,10 +875,16 @@ def reverse(text):
         self.height = height
         self.width = width
         self.favicon_path = favicon_path
-        if self.is_space and enable_queue is None:
+        if enable_queue is not None:
+            self.enable_queue = enable_queue
+            warnings.warn(
+                "The `enable_queue` parameter has been deprecated. Please use the `.queue()` method instead.",
+                DeprecationWarning,
+            )
+        if self.is_space and self.enable_queue is None:
             self.enable_queue = True
-        else:
-            self.enable_queue = enable_queue or False
+        if self.enable_queue is None:
+            self.enable_queue = False
         utils.run_coro_in_background(self.create_limiter, max_threads)
         self.config = self.get_config_file()
         self.share = share
@@ -876,6 +915,15 @@ def reverse(text):
             self.server = server
             self.is_running = True
 
+            if app.blocks.enable_queue:
+                if app.blocks.auth is not None or app.blocks.encrypt:
+                    raise ValueError(
+                        "Cannot queue with encryption or authentication enabled."
+                    )
+                event_queue.Queue.set_url(self.local_url)
+                # Cannot run async functions in background other than app's scope.
+                # Workaround by triggering the app endpoint
+                requests.get(f"{self.local_url}queue/start")
         utils.launch_counter()
 
         # If running in a colab or not able to access localhost,
@@ -973,6 +1021,7 @@ def reverse(text):
             self.block_thread()
         # Block main thread if running in a script to stop script from exiting
         is_in_interactive_mode = bool(getattr(sys, "ps1", sys.flags.interactive))
+
         if not prevent_thread_lock and not is_in_interactive_mode:
             self.block_thread()
 
@@ -1037,6 +1086,8 @@ def close(self, verbose: bool = True) -> None:
         Closes the Interface that was launched and frees the port.
         """
         try:
+            if self.enable_queue:
+                event_queue.Queue.close()
             self.server.close()
             self.is_running = False
             if verbose:
@@ -1054,8 +1105,6 @@ def block_thread(
         except (KeyboardInterrupt, OSError):
             print("Keyboard interruption in main thread... closing server.")
             self.server.close()
-            if self.enable_queue:
-                queueing.close()
 
     def attach_load_events(self):
         """Add a load event for every component whose initial value should be randomized."""