How to handle multi requests to edit image using timbrooks/instruct-pix2pix in Diffusers at the same time?

[alexd725]

Hello, guies.

Im trying to make an api to edit image using instruct-pix2pix in Diffusers.
but when the api receives multi requests, It cant handle multi requests properly.
Even though I tried to make multi objects for multi requests, its not working.

This is my app.py

from flask import Flask, request, jsonify
import os
from dotenv import load_dotenv
from image_processor import ImageProcessor
import threading

app = Flask(__name__)

# Load environment variables
load_dotenv()
STABLE_DIFFUSION_API_KEY = os.getenv("STABLE_DIFFUSION_API_KEY")
DEEPAI_API_KEY = os.getenv("DEEPAI_API_KEY")

# Initialize multiple instances of ImageProcessor
NUM_PROCESSORS = 1
image_processors = [ImageProcessor() for _ in range(NUM_PROCESSORS)]
processor_statuses = [False] * NUM_PROCESSORS

# Define route to handle POST requests for applying effects
@app.route('/apply-effect', methods=['POST'])
def apply_effect_route():
    try:
        # Get data from request
        data = request.json
        src_image_url = data.get('src_image_url')
        effect = data.get('effect')
        model = 'pix'

        # Find an available image processor
        processor_idx = find_available_processor()

        if processor_idx is not None:
            # Mark the selected processor as busy
            processor_statuses[processor_idx] = True

            # Call apply_effect function
            result = image_processors[processor_idx].apply_effect(src_image_url, effect, model)

            # Mark the processor as available
            processor_statuses[processor_idx] = False

            # Return success response with result
            return jsonify({'success': True, 'result': result}), 200
        else:
            return jsonify({'error': 'All processors are busy, please try again later.'}), 503
    except Exception as e:
        # Return error response if an exception occurs
        return jsonify({'error': str(e)}), 500

def find_available_processor():
    for idx, status in enumerate(processor_statuses):
        if not status:
            return idx
    return None

if __name__ == '__main__':
    app.run(debug=True)

This is my image_processor.py

import time
from datetime import datetime
import os
import torch
import replicate
import requests
import effects
import urllib.request
import gc
from PIL import Image, ImageTk
from io import BytesIO
from diffusers import StableDiffusionInstructPix2PixPipeline, EulerAncestralDiscreteScheduler
from diffusers.utils import load_image

gc.collect()
torch.cuda.empty_cache()

class WebImage:
    def __init__(self, url):
        with urllib.request.urlopen(url) as u:
            raw_data = u.read()
        image = Image.open(BytesIO(raw_data))
        image.thumbnail((400, 400))  # Fix here
        self.image = image

    def get(self):
        return self.image

class ImageProcessor:
    def __init__(self):
        self.effects = effects.EFFECTS

        torch.backends.cuda.matmul.allow_tf32 = True
        torch._dynamo.config.suppress_errors = True

        model_id = "timbrooks/instruct-pix2pix"
        self.pipe = StableDiffusionInstructPix2PixPipeline.from_pretrained(model_id, torch_dtype=torch.float16, safety_checker=None)
        self.pipe.to("cuda")
        self.pipe.scheduler = EulerAncestralDiscreteScheduler.from_config(self.pipe.scheduler.config)
        self.pipe.unet = torch.compile(self.pipe.unet, mode="reduce-overhead", fullgraph=True)

    def apply_effect(self, src_image_url, effect, model):
        print(f"=== {effect} start ===")

        begin = time.time()
        generated_result = None  # Initialize result variable

        text_prompt = self.effects[effect]
        # text_prompt = effect

        src_image = WebImage(src_image_url).image

        # Apply effect
        images = self.pipe(text_prompt, image=src_image, num_inference_steps=30).images
        print(f"Edit+image       : {time.time()-begin}")
        image = images[0]

        # Create directory if it doesn't exist
        output_directory = "uploads"
        if not os.path.exists(output_directory):
            os.makedirs(output_directory)
        
        # Save generated image
        current_time = datetime.now()
        filename = current_time.strftime(f"{output_directory}/%Y%m%d%H%M%S") + ".jpg"
        image.save(filename)  
        
        # Assign output to result
        generated_result = filename
        
        return generated_result

Any help would be appreciated. Thanks in advance.

[alexd725]

When the api receives 2 requests at the same time, it processes only one request like the below.

Loading pipeline components...: 100%|███████████████████████████████████████████████████████████████████████████████████| 6/6 [00:01<00:00,  4.22it/s]
Loading pipeline components...: 100%|███████████████████████████████████████████████████████████████████████████████████| 6/6 [00:01<00:00,  4.69it/s]
 * Debugger is active!
 * Debugger PIN: 508-497-715
=== impression_grace start ===
=== impression_grace start ===
100%|█████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 30/30 [00:55<00:00,  1.86s/it]
Edit+image       : 57.91715741157532

[sayakpaul]

Concurrency requires thread-safety which isn't guaranteed in the schedulers of a diffusion system. See relevant discussions here: #7450

[alexd725]

So cant i fix this issue?

[alexd725]

In that case I cant use this model for scalability. is there any solution for scalability? in another workd we can handel more than 2 requests at the same time?