# IO密集型后台任务 - 异步循环
async def io_intensive_background_task():
    """
    IO密集型后台任务 - 以固定间隔运行
    直接使用异步操作，不需要run_in_executor
    """
    logging.info("Starting IO intensive background task")
    while True:
        try:
            logging.info("Running IO intensive task cycle")
            
            # 模拟IO密集型操作，如文件操作或网络请求
            start_time = time.time()
            
            # 模拟多个并发IO操作
            io_tasks = [simulate_io_operation() for _ in range(5)]
            results = await asyncio.gather(*io_tasks)
            
            total_io_time = sum(result['time'] for result in results)
            elapsed = time.time() - start_time
            
            logging.info(f"IO task completed: {len(results)} operations, "
                         f"total IO time: {total_io_time:.2f}s, "
                         f"wall time: {elapsed:.2f}s")
            
            # 等待一段时间再次执行
            await asyncio.sleep(30)  # 每30秒执行一次
            
        except asyncio.CancelledError:
            logging.info("IO intensive background task cancelled")
            break
        except Exception as e:
            logging.error(f"Error in IO intensive task: {e}")
            await asyncio.sleep(5)  # 出错后短暂等待再重试

async def simulate_io_operation():
    """模拟单个IO密集型操作，如网络请求或文件读取"""
    # 模拟IO延迟，如网络延迟
    delay = 0.5 + (1.5 * np.random.random())  # 0.5到2秒的随机延迟
    start_time = time.time()
    
    await asyncio.sleep(delay)  # 模拟IO等待
    
    # 模拟读取的数据
    data_size = int(np.random.random() * 1000)  # 0到1000的随机数
    
    elapsed = time.time() - start_time
    return {
        "time": elapsed,
        "data_size": data_size
    }

@asynccontextmanager
async def lifespan(app: FastAPI):
    """应用生命周期管理"""
    # 记录应用启动
    logging.info("Application starting up")
    
    # 创建进程池
    cpus = get_cpu_limit()
    workers = max(1, cpus - 1)  # 保留一个CPU给系统和事件循环
    app.state.process_pool = ProcessPoolExecutor(max_workers=workers)
    logging.info(f"Process pool created with {workers} workers (detected {cpus} CPUs)")
    
    # 启动后台任务
    app.state.background_tasks = []
    
    # 添加CPU密集型后台任务
    cpu_task = asyncio.create_task(cpu_intensive_background_task(app))
    app.state.background_tasks.append(cpu_task)
    logging.info("CPU intensive background task started")
    
    # 添加IO密集型后台任务
    io_task = asyncio.create_task(io_intensive_background_task())
    app.state.background_tasks.append(io_task)
    logging.info("IO intensive background task started")
    
    yield
    
    # 清理资源
    logging.info("Application shutting down, cleaning up resources")
    
    # 取消所有后台任务
    for task in app.state.background_tasks:
        task.cancel()
    
    if app.state.background_tasks:
        # 等待所有任务正确取消
        await asyncio.gather(*app.state.background_tasks, return_exceptions=True)
        logging.info("All background tasks cancelled")
    
    # 关闭进程池
    app.state.process_pool.shutdown(wait=True)
    logging.info("Process pool shut down")

app = FastAPI(lifespan=lifespan)

@app.get("/status")
async def status():
    return {"status": "ok", "message": "Server is running with background tasks"}

import os
import time
import asyncio
import logging
import numpy as np
from multiprocessing import cpu_count
from concurrent.futures import ProcessPoolExecutor
from fastapi import FastAPI, Request
from contextlib import asynccontextmanager

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)

def get_cpu_limit():
    """获取容器环境中可用的CPU数量"""
    try:
        # 检查cgroups v1
        if os.path.exists("/sys/fs/cgroup/cpu/cpu.cfs_quota_us"):
            with open("/sys/fs/cgroup/cpu/cpu.cfs_quota_us") as fp:
                cfs_quota_us = int(fp.read().strip())
            # 只有当有实际限制时才继续
            if cfs_quota_us > 0:
                with open("/sys/fs/cgroup/cpu/cpu.cfs_period_us") as fp:
                    cfs_period_us = int(fp.read().strip())
                container_cpus = cfs_quota_us // cfs_period_us
                return container_cpus
        
        # 检查cgroups v2
        elif os.path.exists("/sys/fs/cgroup/cpu.max"):
            with open("/sys/fs/cgroup/cpu.max") as fp:
                content = fp.read().strip().split()
                if content[0] != "max":  # 有限制
                    quota = int(content[0])
                    period = int(content[1])
                    container_cpus = quota // period
                    return container_cpus
    
    except Exception as e:
        logging.warning(f"Error detecting container CPU limit: {e}")
    
    # 回退到系统CPU计数
    return cpu_count()

# CPU密集型函数 - 在单独进程中运行
def cpu_intensive_operation(size=1000):
    """
    一个模拟CPU密集型操作的函数，例如矩阵运算
    这个函数在一个单独的进程中运行
    """
    logging.info(f"Starting CPU intensive operation with size {size}")
    # 创建随机矩阵
    matrix_a = np.random.rand(size, size)
    matrix_b = np.random.rand(size, size)
    
    # 执行耗CPU的矩阵乘法
    start_time = time.time()
    result = np.matmul(matrix_a, matrix_b)
    
    # 计算一些统计数据
    mean_value = np.mean(result)
    std_value = np.std(result)
    
    elapsed = time.time() - start_time
    logging.info(f"Completed CPU intensive operation in {elapsed:.2f} seconds")
    
    return {
        "mean": float(mean_value),
        "std": float(std_value),
        "elapsed": elapsed
    }

# CPU密集型后台任务 - 异步循环
async def cpu_intensive_background_task(app):
    """
    CPU密集型后台任务 - 以固定间隔运行
    使用run_in_executor将耗CPU的操作卸载到进程池
    """
    logging.info("Starting CPU intensive background task")
    while True:
        try:
            logging.info("Running CPU intensive task cycle")
            
            # 获取事件循环
            loop = asyncio.get_running_loop()
            
            # 将CPU密集型操作卸载到进程池
            result = await loop.run_in_executor(
                app.state.process_pool, 
                cpu_intensive_operation,  # 传递CPU密集型函数
                500  # 矩阵大小参数
            )
            
            logging.info(f"CPU task result: mean={result['mean']:.4f}, time={result['elapsed']:.2f}s")
            
            # 等待一段时间再次执行
            await asyncio.sleep(60)  # 每分钟执行一次
            
        except asyncio.CancelledError:
            logging.info("CPU intensive background task cancelled")
            break
        except Exception as e:
            logging.error(f"Error in CPU intensive task: {e}")
            await asyncio.sleep(10)  # 出错后短暂等待再重试

# 每个插件必须实现的基础接口
@dataclass
class PluginInfo:
    id: str          # 插件唯一标识
    name: str        # 插件名称
    version: str     # 版本
    description: str # 描述
    author: str      # 作者
    homepage: str    # 可选，项目主页

@dataclass
class ActionParameter:
    type: str
    description: str
    required: bool
    default: Any = None

@dataclass
class Action:
    description: str
    method: str      # GET/POST/PUT/DELETE
    path: str
    params: dict[str, ActionParameter]
    returns: dict    # 返回值描述

@dataclass
class Event:
    description: str
    schema: dict     # JSON Schema 描述事件数据结构

# 基础 HTTP 接口
GET /plugin/info -> PluginInfo
GET /plugin/capabilities -> {
    "actions": dict[str, Action],
    "events": dict[str, Event],
    "subscriptions": list[str]
}
GET /health -> {"status": "ok" | "error"}
POST /events -> 接收订阅的事件

# 下载插件接口定义
POST /actions/download {
    "request": {
        "url": str,
        "save_path": str,
        "headers": dict[str, str]  # 可选
    },
    "response": {
        "task_id": str,
        "status": "started"
    }
}

GET /tasks/{task_id} {
    "response": {
        "task_id": str,
        "status": "downloading" | "completed" | "error",
        "progress": float,  # 0-100
        "speed": int,      # bytes/s
        "downloaded": int,  # bytes
        "total": int       # bytes
    }
}

# 事件定义
events = {
    "download.completed": {
        "task_id": str,
        "file_path": str,
        "file_size": int,
        "mime_type": str
    }
}

sudo dnf install zram-generator-defaults

[zram0]
zram-size = min(ram, 4096)

systemctl restart systemd-zram-setup@zram0