import ray
import time
import os
import sys

if __name__ == "__main__":
    # ✅ 关键修复：设置环境变量以缓解 Windows 端口冲突
    os.environ["RAY_DEDUP_LOGS"] = "0"
    
    try:
        # 连接现有集群
        ray.init(address="auto")
        print("✅ 成功连接到现有 Ray 集群")
    except Exception as e:
        print(f"❌ 连接失败: {e}")
        sys.exit(1)

    def process_data(row):
        val = row.get('item')
        time.sleep(1)
        return {"result": f"Worker(PID:{os.getpid()}) 处理了 {val}"}

    print("🚀 开始处理...")
    ds = ray.data.from_items(list(range(10)))
    
    # ✅ 增加重试机制，避免端口抖动导致任务直接崩溃
    ctx = ray.data.DataContext.get_current()
    ctx.max_errored_blocks = 3  # 允许最多 3 个 Block 失败重试

    try:
        results = ds.map(process_data).take_all()
        print("\n✅ 结果:")
        for r in results:
            print(r)
    except Exception as e:
        print(f"❌ 执行失败: {e}")