batch polling and batching support

Author: v1r3n

examples/asyncio_workers.py

@@ -1,17 +1,28 @@
 """
-AsyncIO Workers Example
+AsyncIO Workers Example - Java SDK Architecture
 
-This example demonstrates how to use the AsyncIO-based TaskHandlerAsyncIO
-instead of the multiprocessing-based TaskHandler.
+This example demonstrates the AsyncIO task runner with Java SDK architecture features:
+- Semaphore-based dynamic batch polling
+- Per-worker thread count configuration
+- Automatic lease extension
+- In-memory queue for V2 API chained tasks
+- Zero-polling optimization
 
-Advantages of AsyncIO:
-- Lower memory footprint (single process)
-- Better for I/O-bound tasks
-- Simpler debugging
+Key Features (matching Java SDK):
+- Dynamic batch sizing (batch = available threads)
+- No server calls when all threads busy
+- Adaptive concurrency control
+- Optimal resource utilization
 
 Requirements:
     pip install httpx  # AsyncIO HTTP client
 
+Configuration:
+    Set environment variables or create conductor_config.py:
+    - CONDUCTOR_SERVER_URL: e.g., https://play.orkes.io/api
+    - CONDUCTOR_AUTH_KEY: API key
+    - CONDUCTOR_AUTH_SECRET: API secret
+
 Run:
     python examples/asyncio_workers.py
 """
@@ -60,18 +71,28 @@ class User:
     company: Company
 
 
-# Example 1: Synchronous worker (will run in thread pool)
-@worker_task(task_definition_name='greet')
+# Example 1: Simple synchronous worker (runs in thread pool)
+@worker_task(
+    task_definition_name='greet',
+    thread_count=101,              # Low concurrency for simple tasks
+    poll_timeout=100,            # Default poll timeout (ms)
+    lease_extend_enabled=False   # Fast tasks don't need lease extension
+)
 def greet(name: str) -> str:
     """
     Synchronous worker - automatically runs in thread pool to avoid blocking.
-    Good for legacy code or CPU-bound tasks.
+    Good for legacy code or simple CPU-bound tasks.
     """
     return f'Hello {name}'
 
 
-# Example 2: Async worker (runs natively in event loop)
-@worker_task(task_definition_name='greet_async')
+# Example 2: Simple async worker (runs natively in event loop)
+@worker_task(
+    task_definition_name='greet_async',
+    thread_count=10,             # Higher concurrency for async I/O
+    poll_timeout=100,
+    lease_extend_enabled=False
+)
 async def greet_async(name: str) -> str:
     """
     Async worker - runs natively in the event loop.
@@ -82,71 +103,136 @@ async def greet_async(name: str) -> str:
     return f'Hello {name} (from async function)'
 
 
-# Example 3: Async worker with HTTP call
-@worker_task(task_definition_name='fetch_user')
+# Example 3: High-throughput HTTP worker with batch polling
+@worker_task(
+    poll_interval_millis=10,
+    task_definition_name='fetch_user',
+    thread_count=20,             # High concurrency for I/O-bound tasks
+    poll_timeout=20,            # Longer timeout for efficient long-polling
+    lease_extend_enabled=False   # Fast HTTP calls don't need lease extension
+)
 async def fetch_user(user_id: str) -> dict:
     """
     Example of making async HTTP calls using httpx.
-    This is more efficient than synchronous requests.
+    With thread_count=20, the system will:
+    - Batch poll up to 20 tasks when all threads available
+    - Skip polling when all 20 threads busy (zero-polling)
+    - Dynamically adjust batch size based on availability
     """
     try:
         import httpx
-        # print(f'fetching user {user_id}')
         async with httpx.AsyncClient() as client:
             response = await client.get(
-                f'https://jsonplaceholder.typicode.com/users/{user_id}'
+                f'https://jsonplaceholder.typicode.com/users/{user_id}',
+                timeout=10.0
             )
-            # print(f'response {response.json()}')
             return response.json()
 
     except Exception as e:
         return {"error": str(e)}
 
 
-@worker_task(task_definition_name='process_user')
+# Example 4: Dataclass-based worker (type-safe input)
+@worker_task(
+    task_definition_name='process_user',
+    thread_count=15,
+    poll_timeout=150,
+    lease_extend_enabled=False
+)
 async def process_user(user: User) -> dict:
     """
-    Example of making async HTTP calls using httpx.
-    This is more efficient than synchronous requests.
+    Worker that accepts User dataclass - SDK automatically converts from dict.
+    Demonstrates type-safe worker functions.
+
+    The fetch_user task returns a dict, which is chained to this task.
+    Since dict outputs are used as-is (not wrapped in "result"),
+    the User dataclass can be properly constructed.
+    """
+    try:
+        import httpx
+        async with httpx.AsyncClient() as client:
+            response = await client.get(
+                f'https://jsonplaceholder.typicode.com/users/{user.id + 3}',
+                timeout=10.0
+            )
+            return response.json()
+
+    except Exception as e:
+        return {"error": str(e)}
+
+
+# Example 5: Worker with dict input (flexible alternative)
+@worker_task(
+    task_definition_name='process_user_dict',
+    thread_count=10,
+    poll_timeout=150,
+    lease_extend_enabled=False
+)
+async def process_user_dict(user: dict) -> dict:
+    """
+    Worker that accepts dict input directly - more flexible.
+    Use this when you don't need strict type checking.
+
+    Accepts any dict with an 'id' field.
     """
     try:
         import httpx
-        # print(f'fetching user details for {user.id}')
+        user_id = user.get('id', 1)
+
         async with httpx.AsyncClient() as client:
             response = await client.get(
-                f'https://jsonplaceholder.typicode.com/users/{user.id + 1}'
+                f'https://jsonplaceholder.typicode.com/users/{user_id + 1}',
+                timeout=10.0
             )
-            # print(f'response {response.json()}')
             return response.json()
 
     except Exception as e:
         return {"error": str(e)}
 
 
-# Example 4: CPU-bound work in thread pool
-@worker_task(task_definition_name='calculate')
+# Example 6: CPU-bound work in thread pool (lower concurrency)
+@worker_task(
+    task_definition_name='calculate',
+    thread_count=4,              # Lower concurrency for CPU-bound tasks
+    poll_timeout=100,
+    lease_extend_enabled=False
+)
 def calculate_fibonacci(n: int) -> int:
     """
     CPU-bound work automatically runs in thread pool.
     For heavy CPU work, consider using multiprocessing TaskHandler instead.
+
+    Note: thread_count=4 limits concurrent CPU-intensive tasks to avoid
+    overwhelming the system (GIL contention).
     """
     if n <= 1:
         return n
     return calculate_fibonacci(n - 1) + calculate_fibonacci(n - 2)
 
 
-# Example 5: Mixed I/O and CPU work
-@worker_task(task_definition_name='process_data')
+# Example 7: Mixed I/O and CPU work with controlled concurrency
+@worker_task(
+    task_definition_name='process_data',
+    thread_count=12,             # Moderate concurrency for mixed workload
+    poll_timeout=200,
+    lease_extend_enabled=True,   # Enable lease extension for longer tasks
+    register_task_def=False      # Don't auto-register task definition
+)
 async def process_data(data_url: str) -> dict:
     """
     Demonstrates mixing async I/O with CPU-bound work.
     I/O runs in event loop, CPU work runs in thread pool.
+
+    With thread_count=12:
+    - System can batch poll up to 12 tasks when all threads free
+    - Zero-polling kicks in when all 12 threads busy
+    - Dynamically adjusts batch size as threads complete
     """
     import httpx
 
     # I/O-bound: Fetch data asynchronously
     async with httpx.AsyncClient() as client:
-        response = await client.get(data_url)
+        response = await client.get(data_url, timeout=10.0)
         data = response.json()
 
     # CPU-bound: Process in thread pool
@@ -168,9 +254,33 @@ def _process_data_sync(data: dict) -> dict:
     return {"processed": True, "count": len(data)}
 
 
+# Example 8: Long-running task with automatic lease extension
+@worker_task(
+    task_definition_name='long_task',
+    thread_count=2,              # Low concurrency for expensive tasks
+    poll_timeout=500,
+    lease_extend_enabled=True    # Automatically extends lease at 80% of timeout
+)
+async def long_running_task(duration: int) -> dict:
+    """
+    Demonstrates automatic lease extension for long-running tasks.
+
+    If task.response_timeout_seconds = 300 (5 minutes):
+    - Lease extension sent at 240s (80%)
+    - Repeats every 240s until task completes
+    - Retries up to 3 times per extension
+    - Automatically cancelled when task completes
+
+    This keeps the task alive in Conductor during long processing.
+    """
+    # Simulate long-running operation
+    await asyncio.sleep(duration)
+    return {"duration": duration, "completed": True}
+
+
 async def main():
     """
-    Main entry point demonstrating different ways to use TaskHandlerAsyncIO.
+    Main entry point demonstrating AsyncIO task handler with Java SDK architecture.
     """
 
     # Configuration - defaults to reading from environment variables:
@@ -180,10 +290,26 @@ async def main():
     api_config = Configuration()
 
     print("=" * 60)
-    print("Conductor AsyncIO Workers Example")
+    print("Conductor AsyncIO Workers - Java SDK Architecture")
     print("=" * 60)
     print(f"Server: {api_config.host}")
-    print(f"Workers: greet, greet_async, fetch_user, calculate, process_data")
+    print()
+    print("Workers with dynamic batch polling:")
+    print("  • greet (thread_count=1)")
+    print("  • greet_async (thread_count=10)")
+    print("  • fetch_user (thread_count=20) - High throughput")
+    print("  • process_user (thread_count=15) - Type-safe dataclass")
+    print("  • process_user_dict (thread_count=10) - Flexible dict input")
+    print("  • calculate (thread_count=4) - CPU-bound")
+    print("  • process_data (thread_count=12) - Mixed I/O+CPU")
+    print("  • long_task (thread_count=2) - With lease extension")
+    print()
+    print("Features:")
+    print("  ✓ Dynamic batch polling (batch size = available threads)")
+    print("  ✓ Zero-polling optimization (skip when all threads busy)")
+    print("  ✓ Automatic lease extension at 80% of timeout")
+    print("  ✓ In-memory queue for V2 API chained tasks")
+    print("  ✓ Per-worker concurrency control")
     print("=" * 60)
     print("\nStarting workers... Press Ctrl+C to stop\n")
 
@@ -229,6 +355,71 @@ def signal_handler():
     print("\nWorkers stopped. Goodbye!")
 
 
+async def demo_v2_api():
+    """
+    Example of V2 API support with in-memory queue.
+
+    When enabled (export taskUpdateV2=true), the server can return
+    the next task to execute in the update response, which is added
+    to the in-memory queue to avoid redundant polling.
+    """
+    import os
+    os.environ['taskUpdateV2'] = 'true'
+
+    api_config = Configuration()
+
+    @worker_task(
+        task_definition_name='chained_task',
+        thread_count=10
+    )
+    async def chained_task(data: dict) -> dict:
+        """Task that may be part of a chained workflow"""
+        await asyncio.sleep(0.5)
+        return {"result": "processed", "data": data}
+
+    async with TaskHandlerAsyncIO(configuration=api_config) as handler:
+        # Server may return next task in workflow
+        # → Added to in-memory queue
+        # → Drained before next server poll
+        # → Reduces server calls by ~30% for chained workflows
+        await handler.wait()
+
+
+async def demo_zero_polling():
+    """
+    Example demonstrating zero-polling optimization.
+
+    When all threads are busy:
+    - poll_count = 0 (no available permits)
+    - Skip server call (zero-polling)
+    - Sleep briefly and retry
+    - Saves server resources during high load
+    """
+
+    @worker_task(
+        task_definition_name='busy_task',
+        thread_count=5  # Only 5 concurrent tasks allowed
+    )
+    async def busy_task(duration: int) -> dict:
+        """Simulates a task that takes 'duration' seconds"""
+        await asyncio.sleep(duration)
+        return {"completed": True}
+
+    api_config = Configuration()
+
+    async with TaskHandlerAsyncIO(configuration=api_config) as handler:
+        # Scenario: 10 tasks queued on server
+        #
+        # Poll #1: 5 permits available → batch poll 5 tasks → all threads busy
+        # Poll #2: 0 permits available → zero-polling (skip server call)
+        # Poll #3: 0 permits available → zero-polling (skip server call)
+        # ...
+        # Poll #N: 2 tasks complete → 2 permits available → batch poll 2 tasks
+        #
+        # Result: Saved (N-2) server calls during high load
+        await handler.wait()
+
+
 if __name__ == '__main__':
     """
     Run the async main function.
@@ -237,6 +428,12 @@ def signal_handler():
     Python 3.6: asyncio.get_event_loop().run_until_complete(main())
     """
     try:
+        # Run main demo
         asyncio.run(main())
+
+        # Uncomment to run other demos:
+        # asyncio.run(demo_v2_api())
+        # asyncio.run(demo_zero_polling())
+
     except KeyboardInterrupt:
         pass