RunpodR
Runpod•2y ago•
88 replies
J.

Does async generator allow a worker to take off multiple jobs? Concurrency Modifier?

I was reading the runpod docs, and I saw the below. But does an async generator_handler, mean that if I sent 5 jobs for example that one worker will just keep on picking up new jobs?

I also tried to add the:
"concurrency_modifier": 4


But if I queued up 10 jobs, it would first max out the workers, and just have jobs sitting in queue, rather than each worker picking up to the max number of concurrency modifiers?

https://docs.runpod.io/serverless/workers/handlers/handler-async
import runpod
import asyncio

async def async_generator_handler(job):
    for i in range(5):
        # Generate an asynchronous output token
        output = f"Generated async token output {i}"
        yield output

        # Simulate an asynchronous task, such as processing time for a large language model
        await asyncio.sleep(1)


# Configure and start the RunPod serverless function
runpod.serverless.start(
    {
        "handler": async_generator_handler,  # Required: Specify the async handler
        "return_aggregate_stream": True,  # Optional: Aggregate results are accessible via /run endpoint
    }
)
Asynchronous Handler | RunPod Documentation
RunPod supports the use of asynchronous handlers, enabling efficient handling of tasks that benefit from non-blocking operations. This feature is particularly useful for tasks like processing large datasets, interacting with APIs, or handling I/O-bound operations.
Asynchronous Handler | RunPod Documentation
Solution
import runpod
import asyncio
import random


async def process_request(job):
    await asyncio.sleep(10)  # Simulate processing time
    return f"Processed: {job['data']}"

def adjust_concurrency(current_concurrency):
    """
    Adjusts the concurrency level based on the current request rate.
    """
    return 5

# Start the serverless function with the handler and concurrency modifier
runpod.serverless.start({
    "handler": process_request,
    "concurrency_modifier": adjust_concurrency
})
Jump to solution
Was this page helpful?