在Flask应用中高效处理GPU密集型后台任务

import time

def apply_algorithm(file):
    """
    模拟GPU密集型算法，耗时约70秒。
    """
    print(f"开始处理文件: {file}")
    time.sleep(70)  # 模拟耗时操作
    print(f"文件 {file} 处理完成")
    return f"Processed {file}"

# server.py (modified)
import json, logging
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
from flask import Flask, request
import time # 用于模拟任务

logging.basicConfig(format='[%(asctime)s] %(message)s', datefmt='%Y-%m-%d %H:%M:%S', level=logging.INFO)

app = Flask(__name__)

# 根据任务类型选择合适的执行器
# 对于GPU任务，通常会释放GIL，ThreadPoolExecutor可能足够
# 但如果任务包含CPU密集型预处理或后处理，ProcessPoolExecutor更佳
EXECUTOR = ProcessPoolExecutor(max_workers=4) # 可以指定工作进程/线程数量

def apply_algorithm(file):
    # 模拟GPU相关算法 -- 图像/视频分析 (非常耗时的任务)
    print(f"[{time.ctime()}] 开始处理文件: {file}")
    time.sleep(70) # 模拟GPU任务耗时
    print(f"[{time.ctime()}] 文件 {file} 处理完成")
    return f"Analysis complete for {file}"

@app.route('/analyze', methods = ['POST'])
def analyze():
    file = request.form['file']
    message = None

    try:
        # 提交任务到后台执行器，立即返回Future对象
        EXECUTOR.submit(apply_algorithm, file) 
        message = f'Processing started for {file}!'
        logging.info(message)
    except Exception as error:
        message = f'Error: Unable to analyze {file}!'
        logging.warning(f"Error submitting task for {file}: {error}")

    status = {'status': message}

    # 立即返回响应给客户端
    return json.dumps(status)

if __name__ == "__main__":
    # 启用多线程模式，允许服务器同时处理多个请求
    app.run(debug=True, host='0.0.0.0', port=5000, threaded=True)

import requests
import time

def send_request(host, port, file):
    url = f'http://{host}:{port}/analyze'
    body = {'file': file}
    print(f"[{time.ctime()}] Sending request for {file}...")
    try:
        response = requests.post(url, data = body, timeout=5) # 设置一个较短的超时，因为服务器应立即响应
        status = response.json()['status']
        print(f"[{time.ctime()}] Received response for {file}: {status}")
    except requests.exceptions.Timeout:
        print(f"[{time.ctime()}] Request for {file} timed out, but server should have responded.")
    except Exception as e:
        print(f"[{time.ctime()}] Error sending request for {file}: {e}")


if __name__ == "__main__":
    # 模拟多个客户端并发请求
    import threading
    files = ["test1.h5", "test2.h5", "test3.h5"]
    threads = []
    for f in files:
        t = threading.Thread(target=send_request, args=("localhost", 5000, f))
        threads.append(t)
        t.start()
        time.sleep(0.1) # 稍微错开请求时间

    for t in threads:
        t.join()
    print(f"[{time.ctime()}] All client requests sent and processed (or timed out).")

gunicorn -w 4 -k gevent -b 0.0.0.0:5000 server:app

import json, logging
from concurrent.futures import ProcessPoolExecutor
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import socketserver
import time # 用于模拟任务

logging.basicConfig(format='[%(asctime)s] %(message)s', datefmt='%Y-%m-%d %H:%M:%S', level=logging.INFO)

# 设置TCP服务器的请求队列大小，防止连接被拒绝
socketserver.TCPServer.request_queue_size = 100

EXECUTOR = ProcessPoolExecutor(max_workers=4)

def apply_algorithm(file):
    print(f"[{time.ctime()}] 开始处理文件 (ThreadingHTTPServer): {file}")
    time.sleep(70) # 模拟GPU任务耗时
    print(f"[{time.ctime()}] 文件 {file} 处理完成 (ThreadingHTTPServer)")
    return f"Analysis complete for {file}"

class FunctionServerHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        content_len = int(self.headers.get('Content-Length'))
        post_body = self.rfile.read(content_len)
        data = json.loads(post_body.decode('utf-8'))
        file = data.get("file")

        try:
            # 提交任务到后台执行器，并立即返回响应
            EXECUTOR.submit(apply_algorithm, file)
            message = f'Processing started for {file}!'
            self.send_response(200)
            self.send_header('Content-type', 'application/json')
            self.end_headers()
            self.wfile.write(json.dumps({"status": message}).encode('utf-8'))
            self.wfile.flush()
            logging.info(message)
        except Exception as error:
            message = f'Error: Unable to analyze {file}!'
            logging.warning(f"Error submitting task for {file}: {error}")
            self.send_response(500)
            self.send_header('Content-type', 'application/json')
            self.end_headers()
            self.wfile.write(json.dumps({"status": message}).encode('utf-8'))
            self.wfile.flush()

    def log_message(self, format, *args):
        # 禁用默认的HTTP请求日志，以免与自定义日志混淆
        return

if __name__ == "__main__":
    host = "0.0.0.0"
    port = 5000
    print(f"[{time.ctime()}] Starting ThreadingHTTPServer on {host}:{port}")
    httpd = ThreadingHTTPServer((host, port), FunctionServerHandler)
    httpd.serve_forever()