榨干云上 I/O 与带宽：基于华为云鲲鹏（Kunpeng）架构与 OBS 构建高吞吐自动化数据流转引擎

# 测试 128K 块大小下的顺序混合读写吞吐量（模拟大文件流转缓冲）
fio --name=essd_seq_test \
    --filename=/mnt/data/fio_test.bin \
    --ioengine=io_uring \
    --direct=1 \
    --rw=rw --rwmixread=50 \
    --bs=128k \
    --iodepth=64 \
    --numjobs=4 \
    --size=10G \
    --runtime=60 --time_based --group_reporting

# 服务端：iperf3 -s
# 客户端：开启 8 个并行流，持续压测 30 秒
iperf3 -c 192.168.1.102 -P 8 -t 30 -O 3

import asyncio
import aiohttp
import logging
from pathlib import Path

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")

class OBSTransferScheduler:
    def __init__(self, rc_endpoint: str, obs_bucket: str, max_concurrency: int = 8):
        self.rc_endpoint = rc_endpoint
        self.obs_bucket = obs_bucket
        # 使用信号量严格控制并发任务数，防止打爆本地 IOPS
        self.semaphore = asyncio.Semaphore(max_concurrency)

    async def transfer_file(self, session: aiohttp.ClientSession, local_path: Path, remote_prefix: str):
        async with self.semaphore:
            dst_path = f"huaweicloud-obs:{self.obs_bucket}/{remote_prefix}/{local_path.name}"
            payload = {
                "srcFs": str(local_path.parent),
                "srcRemote": local_path.name,
                "dstFs": f"huaweicloud-obs:{self.obs_bucket}/{remote_prefix}",
                "dstRemote": local_path.name,
                "_async": True,
                "_config": {
                    "Transfers": 16,
                    "S3UploadConcurrency": 8,
                    "S3ChunkSize": "16M",  # 对齐大块顺序读取，最大化利用 ESSD 吞吐
                    "BufferSize": "32M"
                }
            }
            
            for attempt in range(1, 4):
                try:
                    async with session.post(f"{self.rc_endpoint}/operations/copyfile", json=payload) as resp:
                        if resp.status == 200:
                            data = await resp.json()
                            job_id = data.get("jobid")
                            logging.info(f"Task dispatched [JobID: {job_id}]: {local_path.name} -> {dst_path}")
                            await self._wait_and_cleanup(session, job_id, local_path)
                            return
                        else:
                            raise RuntimeError(f"HTTP {resp.status}: {await resp.text()}")
                except Exception as e:
                    backoff = 2 ** attempt
                    logging.warning(f"Attempt {attempt} failed for {local_path.name}: {e}. Retrying in {backoff}s...")
                    await asyncio.sleep(backoff)

    async def _wait_and_cleanup(self, session: aiohttp.ClientSession, job_id: int, local_path: Path):
        """轮询异步任务状态，确认云端 OBS 校验落盘后安全释放本地 SSD 空间"""
        while True:
            async with session.post(f"{self.rc_endpoint}/job/status", json={"jobid": job_id}) as resp:
                status = await resp.json()
                if status.get("finished"):
                    if status.get("success"):
                        local_path.unlink(missing_ok=True)
                        logging.info(f"Job {job_id} completed. Local cache freed: {local_path.name}")
                    else:
                        logging.error(f"Job {job_id} failed: {status.get('error')}")
                    break
            await asyncio.sleep(1.5)

# /etc/systemd/system/obs-flow-engine.service
[Unit]
Description=High-Throughput Huawei Cloud OBS Data Flow Engine
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
User=flowrunner
WorkingDirectory=/opt/obs-flow
ExecStart=/opt/obs-flow/.venv/bin/python main.py
Restart=always
RestartSec=3

# 结合 cgroups v2 限制峰值内存，防止大文件缓冲区失控触发系统级 OOM Killer
MemoryHigh=4G
MemoryMax=6G

# 提升 I/O 调度优先级与文件描述符上限
IOSchedulingClass=best-effort
IOSchedulingPriority=2
LimitNOFILE=1048576

[Install]
WantedBy=multi-user.target