How to ensure requests go to multiple Runpod workers at once?

[aleph65]

Sounds like a silly question

But when I have 10 runpod workers going, for some weird reason only 1 of them keeps getting my Fooocus API requests, even though locally I am sending a bunch of requests at once with runsync (precisely 10 at a time).

Any suggestions on how the workers distributes the requests, and best practices for a high success rate in the requests? Like when is the best time / way to send a follow up request if a worker is processing something? e.g. do I have to check status or something like that...

Thank you and sorry for the somewhat open ended questions

[davefojtik]

Do you see them in queue and running one by one in RunPod dashboard or does it register just one request and others are discarded?

[davefojtik]

To the second part of the question - RunPod can work in a way that as long as you have available workers, it takes the next request and process it. If no free workers are available it puts it into a queue that gets processed once you have free workers again.

To ensure this behaviour, double check your serverless settings, especially Advanced Settings -> Scale Type

[aleph65]

Thank you so much for the quick answer. I'm going to log the failed requests to understand what's returned

I was running a large batch and all I saw was that no "output" was returned and had to retry until an image was returned...

I had modified my scale type with the intent to always start a new worker as soon as possible. But if I warm them up before hand (set Active to 10), does scale type actually do anything / influence the routing?

[davefojtik]

It should work the same even with Active Workers. But you can try to turn them off to be sure. You can also try setting the scaling to Queue Delay - that is time-based. Normally once you have the workers in the Idle state they can receive new requests.

As you send the requests you should see the workers taking them or waiting in a queue

You should also see them in the Requests tab below this

To troubleshoot this better and determine if there's a problem with the RunPod or the way you're sending the requests, please also try these things:

• send the requests with something like Postman or similar tools.
• send them to the async /run endpoint and see if there's a difference.
• make sure your environment sending the requests utilizes asynchronous processing or multithreading, allowing you to process sync requests concurrently and not only sequentially. (See Async vs Sync requests)

Let me know the results. Also feel free to share more about your app, environment or outputs you're receiving so I can assist better.

[aleph65]

Thank you SO much for the suggestions. I'll try doing the async endpoint instead, maybe that can help clean things up. I will also revisit my parallel local batch processing. Here's my batch processing which only copied from ChatGPT, I'm not sure how the parallelism here works (or if it's actually parallel at all or single threaded still)

MAX_CONCURRENT_REQUESTS = 26

async def process_image_directory(directory_path):
    output_folder = create_output_folder("ppv_batcher/ppv-batcher")
    valid_extensions = {".jpg", ".jpeg", ".png", ".gif", ".bmp"}
    image_files = sorted(
        f
        for f in os.listdir(directory_path)
        if os.path.isfile(os.path.join(directory_path, f))
        and os.path.splitext(f)[1].lower() in valid_extensions
    )

    semaphore = asyncio.Semaphore(MAX_CONCURRENT_REQUESTS)

    async def process_single_image(sem, filename, index, folder):
        async with (
            sem
        ):  # This will block if `MAX_CONCURRENT_REQUESTS` are already running
            original_path = os.path.join(directory_path, filename)
            extension = os.path.splitext(filename)[1].lower()
            new_filename = f"{index:04d}{extension}"
            copied_path = os.path.join(folder, new_filename)
            shutil.copyfile(original_path, copied_path)
            await generate_all_variations_for_image(
                copied_path, index, folder, len(image_files)
            )

    # Create a list of task coroutines, each wrapped with the semaphore for concurrency control
    tasks = [
        process_single_image(semaphore, filename, index, output_folder)
        for index, filename in enumerate(image_files, start=1)
    ]

    # Use asyncio.gather to run tasks concurrently, respecting the semaphore limit
    await asyncio.gather(*tasks)


async def main(image_path_or_directory):
    if os.path.isfile(image_path_or_directory):
        await process_image_file(image_path_or_directory)
    elif os.path.isdir(image_path_or_directory):
        await process_image_directory(image_path_or_directory)
    else:
        print(
            "The path provided does not exist or is not recognized as a file or directory."
        )


if __name__ == "__main__":
    if len(sys.argv) != 2:
        print("Usage: python script.py PATH_TO_IMAGE_OR_DIRECTORY")
        sys.exit(1)

    path = sys.argv[1]
    asyncio.run(main(path))

[davefojtik]

Thank you for sharing more details about the enviroment. It seems like you want to generate variations of images from a local folder. But the generate_all_variations_for_image function is not defined in the snippet so I still don't know how you're sending requests to your endpoint, thus I can't directly assess your method. I would also discourage you from copying and using a code from chatGPT without understanding what it does. AI, like the fire, is a great tool but a very bad master.

Anyway, sending batch sync requests in Python could look something like this:

import concurrent.futures
import requests
import json

num = 5 # Number of available workers
url = "https://api.runpod.ai/v2/endpoint-id/runsync"
data = {"input":{"api_name":"txt2img","prompt":"your-prompt","require_base64":True}}
headers = {'Content-Type':'application/json', 'Authorization':'Bearer YOURTOKEN'}

def send_request():
    response = requests.post(url, data=json.dumps(data), headers=headers)
    print(response.json())
    
with concurrent.futures.ThreadPoolExecutor() as executor: 
    futures = [executor.submit(send_request) for _ in range(num)] 
        
concurrent.futures.wait(futures)

The above code works just fine, so try to revise yours accordingly. Don't hesitate to ask if you'll need more help.