## lf001主节点
export VLLM_HOST_IP=lf001
ray start --head \
  --node-ip-address=lf001 \
  --port=6379 \
  --include-dashboard=True \
  --dashboard-host=0.0.0.0 \
  --dashboard-port=8265 \
  --num-gpus=1 \
  --resources='{"node:lf001": 1}'

##lf002 节点
export VLLM_HOST_IP=lf002
ray start --address="lf001:6379" \
  --node-ip-address=lf002 \
  --num-gpus=1 \
  --resources='{"node:lf002": 1}'

　# 查看状态
　ray status

## 查看节点状态
ray list nodes

#!/bin/bash
echo $GLOO_SOCKET_IFNAME
export NCCL_IB_DISABLE=1
export NCCL_SOCKET_IFNAME=eth0
export GLOO_SOCKET_IFNAME=eth0
export NCCL_DEBUG=DEBUG
export VLLM_LOGGING_LEVEL=DEBUG

export VLLM_DISTRIBUTED_BACKEND=nccl
export TORCH_DISTRIBUTED_BACKEND=nccl
export RAY_ADDRESS=10.60.182.19:6379

export VLLM_USE_RAY_SPMD_WORKER=0
export TORCH_DISTRIBUTED_INIT_TIMEOUT=600
export GLOO_TIMEOUT=600

python -m vllm.entrypoints.openai.api_server \
  --model /opt/models/ \
  --tensor-parallel-size 2 \
  --pipeline-parallel-size 1 \
  --trust-remote-code \
  --gpu-memory-utilization 0.75 \
  --max-num-seqs 64 \
  --max-model-len 4096 \
  --host 0.0.0.0 \
  --port 8000 \
  --distributed-executor-backend=ray

import ray
import time

# 1. 初始化 Ray（单机环境，自动启动本地 Ray 集群；多节点环境指定 address 即可）
ray.init(ignore_reinit_error=True)  # ignore_reinit_error 避免重复初始化报错

# 2. 定义并行任务（用 @ray.remote 装饰，标记为 Ray 远程任务）
# 模拟一个耗时算法（添加 1 秒延时，体现并行优势）
@ray.remote
def calculate_square(num: int) -> int:
    """计算数字的平方，模拟耗时算法任务"""
    time.sleep(1)  # 模拟算法执行耗时（如数据处理、模型推理）
    return num * num

# 3. 串行执行（作为对比，查看耗时）
print("=== 串行执行 ===")
start_time = time.time()
serial_results = [calculate_square(num) for num in range(10)]  # 直接调用函数（串行）
serial_cost = time.time() - start_time
print(f"串行结果：{serial_results}")
print(f"串行耗时：{serial_cost:.2f} 秒\n")

# 4. 并行执行（使用 Ray Task 分布式调度）
print("=== 并行执行 ===")
start_time = time.time()
# 4.1 提交所有远程任务（返回任务 ID，不阻塞）
remote_tasks = [calculate_square.remote(num) for num in range(10)]
# 4.2 获取所有任务结果（阻塞，直到所有任务完成）
parallel_results = ray.get(remote_tasks)
parallel_cost = time.time() - start_time
print(f"并行结果：{parallel_results}")
print(f"并行耗时：{parallel_cost:.2f} 秒")

# 5. 关闭 Ray 集群（可选，单机环境运行完毕后可关闭）
ray.shutdown()

import ray
import time

# 1. 初始化 Ray
ray.init(ignore_reinit_error=True)

# 2. 定义有状态 Actor 类（用 @ray.remote 装饰，标记为 Ray Actor）
@ray.remote
class CounterActor:
    """有状态计数器，模拟累积算法结果的有状态任务"""
    def __init__(self, name: str):
        # 初始化状态（如模型参数、累积指标、历史记录等）
        self.name = name
        self.count = 0  # 核心状态：计数器值
    
    def increment(self, step: int = 1) -> int:
        """增加计数器值，模拟算法更新状态"""
        time.sleep(0.5)  # 模拟状态更新耗时（如模型参数更新）
        self.count += step
        return self.count
    
    def get_count(self) -> tuple:
        """获取当前计数器状态，模拟获取算法累积结果"""
        return (self.name, self.count)

# 3. 创建 3 个 Actor 实例（并行运行，各自维护独立状态）
actor1 = CounterActor.remote("计数器1")
actor2 = CounterActor.remote("计数器2")
actor3 = CounterActor.remote("计数器3")

# 4. 并行调用 Actor 方法（更新状态，无阻塞提交任务）
print("=== 并行更新 Actor 状态 ===")
start_time = time.time()
# 每个 Actor 执行 5 次增量操作
remote_tasks = []
for _ in range(5):
    remote_tasks.append(actor1.increment.remote(2))  # 每次加 2
    remote_tasks.append(actor2.increment.remote(3))  # 每次加 3
    remote_tasks.append(actor3.increment.remote(1))  # 每次加 1

# 等待所有更新任务完成
ray.get(remote_tasks)

# 5. 获取所有 Actor 的最终状态
actor_results = ray.get([
    actor1.get_count.remote(),
    actor2.get_count.remote(),
    actor3.get_count.remote()
])

# 6. 输出结果和耗时
cost_time = time.time() - start_time
print(f"所有 Actor 最终状态：{actor_results}")
print(f"并行更新耗时：{cost_time:.2f} 秒")

# 7. 关闭 Ray 集群
ray.shutdown()

import ray
import time
import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split

# 1. 初始化 Ray
ray.init(ignore_reinit_error=True)

# 2. 准备数据和训练模型（模拟已训练好的业务模型）
print("=== 准备数据和训练模型 ===")
X, y = make_classification(n_samples=1000, n_features=20, random_state=42)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.8, random_state=42)

# 训练 RandomForest 模型
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train, y_train)
print(f"模型训练完成，测试数据量：{len(X_test)} 条\n")

# 3. 定义并行预测任务（拆分批量数据，独立预测）
@ray.remote
def batch_predict(model, X_batch: np.ndarray) -> np.ndarray:
    """批量数据预测，模拟机器学习算法推理"""
    time.sleep(0.5)  # 模拟推理耗时（如复杂模型预测）
    return model.predict(X_batch)

# 4. 串行预测（对比耗时）
print("=== 串行预测 ===")
start_time = time.time()
serial_predictions = model.predict(X_test)
serial_cost = time.time() - start_time
print(f"串行预测完成，结果长度：{len(serial_predictions)}")
print(f"串行耗时：{serial_cost:.2f} 秒\n")

# 5. 并行预测（拆分数据为多个批次，并行调度）
print("=== 并行预测 ===")
start_time = time.time()
# 5.1 拆分测试数据为 8 个批次（按需调整批次数量，适配CPU核心数）
batch_size = len(X_test) // 8
X_batches = [X_test[i*batch_size : (i+1)*batch_size] for i in range(8)]
# 补充最后一个批次的剩余数据
if len(X_test) % 8 != 0:
    X_batches.append(X_test[8*batch_size:])

# 5.2 提交并行预测任务
remote_tasks = [batch_predict.remote(model, batch) for batch in X_batches]

# 5.3 获取并合并预测结果
parallel_batch_results = ray.get(remote_tasks)
parallel_predictions = np.concatenate(parallel_batch_results)

# 5.4 输出结果和耗时
parallel_cost = time.time() - start_time
print(f"并行预测完成，结果长度：{len(parallel_predictions)}")
print(f"并行耗时：{parallel_cost:.2f} 秒")
print(f"串行与并行结果是否一致：{np.array_equal(serial_predictions, parallel_predictions)}")

# 6. 关闭 Ray 集群
ray.shutdown()