Kubernetes集群中GPU共享调度与拓扑感知优化实战

通信效率 = Σ(传输量_i × 带宽_i) / 总传输量
拓扑损失率 = 1 - (实际通信效率 / 理论峰值效率)

// 共享GPU设备插件状态机
type GPUPluginState int

const (
    StateInit GPUPluginState = iota
    StateDiscovering
    StateAllocating
    StateReleasing
)

// 拓扑感知设备管理器
type TopologyAwareManager struct {
    nodes         map[string]*GPUNode // 节点拓扑信息
    allocationMap map[string]*PodAlloc // Pod分配记录
    healthChecker *HealthMonitor      // 设备健康监测
}

// GPU节点拓扑结构
type GPUNode struct {
    uuid      string
    topology  GPUTopology
    freeMem   map[int]uint64 // GPU索引->可用显存
    nvLinks   map[int][]int  // GPU索引->连接设备
}

// 拓扑信息采集
func (m *TopologyAwareManager) discoverTopology() error {
    cmd := exec.Command("nvidia-smi", "topo", "-m", "-j")
    output, err := cmd.CombinedOutput()
    if err != nil {
        return fmt.Errorf("topology discovery failed: %v", err)
    }
    
    var topo TopologyJSON
    if err := json.Unmarshal(output, &topo); err != nil {
        return err
    }
    
    // 构建拓扑图
    for _, gpu := range topo.GPUS {
        node := &GPUNode{
            uuid:     gpu.UUID,
            topology: make(map[int]LinkInfo),
        }
        for _, link := range gpu.Links {
            node.topology[link.PeerGPU] = LinkInfo{
                Type:      link.Type,
                Bandwidth: link.Bandwidth,
            }
        }
        m.nodes[gpu.UUID] = node
    }
    return nil
}

class TopologyScorer:
    def __init__(self, topology_map):
        self.topology = topology_map
    
    def score_node(self, pod_req, node_status):
        """
        pod_req: {
            "gpu_count": 4,
            "min_bandwidth": 50 # GB/s
        }
        node_status: {
            "gpus": [
                {"id":0, "free_mem":40960},
                {"id":1, "free_mem":20480},
                ...
            ],
            "topology": [[0,1,50], [0,2,25]] # [gpuA, gpuB, bandwidth]
        }
        """
        # 候选分组算法
        candidate_groups = self.find_contiguous_groups(
            node_status["gpus"], 
            pod_req["gpu_count"],
            pod_req["min_bandwidth"]
        )
        
        if not candidate_groups:
            return 0  # 不满足条件
        
        # 选择最优分组
        best_group = max(candidate_groups, key=lambda g: self.group_score(g))
        return self.group_score(best_group)
    
    def find_contiguous_groups(self, gpus, count, min_bw):
        """BFS搜索满足带宽的连续GPU组"""
        graph = self.build_graph(min_bw)
        groups = []
        
        for start_gpu in graph.nodes:
            visited = set()
            queue = deque([(start_gpu, [start_gpu])])
            
            while queue:
                current, path = queue.popleft()
                if len(path) == count:
                    groups.append(path)
                    continue
                
                for neighbor in graph.neighbors(current):
                    if neighbor not in visited:
                        visited.add(neighbor)
                        new_path = path + [neighbor]
                        queue.append((neighbor, new_path))
        return groups

    def group_score(self, group):
        """计算分组质量得分"""
        intra_bandwidth = 0
        for i in range(len(group)):
            for j in range(i+1, len(group)):
                intra_bandwidth += self.topology.get_bandwidth(group[i], group[j])
        
        return intra_bandwidth / (len(group)*(len(group)-1)/2)

// 内核模块显存隔离核心逻辑
static int gpu_mem_cgroup_alloc(struct cgroup_subsys_state *css)
{
    struct gpu_mem_cgroup *gmem = css_to_gmem(css);
    
    // 创建显存控制结构
    gmem->limits = kzalloc(sizeof(struct gpu_mem_limit), GFP_KERNEL);
    if (!gmem->limits)
        return -ENOMEM;
    
    // 初始化默认限制
    gmem->limits->mem_limit = GPU_MEM_DEFAULT_LIMIT;
    atomic_set(&gmem->limits->mem_usage, 0);
    
    return 0;
}

// 显存分配拦截
static bool gpu_mem_try_charge(size_t size, struct gpu_mem_cgroup *gmem)
{
    long new_usage = atomic_long_add_return(size, &gmem->limits->mem_usage);
    
    if (new_usage > gmem->limits->mem_limit) {
        atomic_long_sub(size, &gmem->limits->mem_usage);
        return false; // 超出限额
    }
    return true;
}

// 驱动层拦截点
int nvidia_mem_alloc_hook(struct nvidia_device *dev, size_t size)
{
    struct gpu_mem_cgroup *gmem = current->gmem_cgroup;
    
    if (!gpu_mem_try_charge(size, gmem)) {
        printk(KERN_WARNING "GPU mem overlimit: pid=%d usage=%ld limit=%ld\n",
               current->pid, atomic_long_read(&gmem->limits->mem_usage),
               gmem->limits->mem_limit);
        return -ENOMEM;
    }
    return 0;
}

# 显存压力测试脚本
for frac in 0.3 0.5 0.8 1.0 1.2; do
    kubectl exec -it $POD -- python -c "
import tensorflow as tf
try:
    # 尝试分配超标显存
    bytes = int($frac * ${GPU_MEM_TOTAL})
    dummy = tf.ones([bytes//4], dtype=tf.float32)
    print(f'Allocated {bytes//1024**2}MB successfully')
except Exception as e:
    print(f'Allocation failed: {str(e)}')
"
done

def comprehensive_scoring(group, pod_req):
    """综合评分模型"""
    # 基础带宽分
    bandwidth_score = calc_bandwidth_score(group) 
    
    # 拓扑结构分（全连接/环状/树状）
    topology_score = calc_topology_structure_score(group)
    
    # 资源平衡分
    balance_score = 1 - (max_mem_util(group) - min_mem_util(group))
    
    # 故障域分散分
    domain_score = calc_fault_domain_dispersion(group)
    
    weights = {
        'bandwidth': 0.5,
        'topology': 0.3,
        'balance': 0.1,
        'domain': 0.1
    }
    
    final_score = (
        weights['bandwidth'] * bandwidth_score +
        weights['topology'] * topology_score +
        weights['balance'] * balance_score +
        weights['domain'] * domain_score
    )
    return final_score

bar
    title GPU利用率对比(%)
    x-axis 模式
    y-axis 利用率 0 100
    
    section 视觉任务
    独占模式 : 22
    基础共享 : 58
    拓扑优化 : 89
    
    section NLP任务
    独占模式 : 18
    基础共享 : 63
    拓扑优化 : 91
    
    section 科学计算
    独占模式 : 25
    基础共享 : 52
    拓扑优化 : 84

-- 拓扑匹配度监控
SELECT
  job_name,
  AVG(topology_score) as avg_score,
  PERCENTILE(topology_score, 0.5) as p50,
  PERCENTILE(topology_score, 0.9) as p90
FROM gpu_scheduling_metrics
WHERE topology_score < 80  -- 低于阈值
GROUP BY job_name
HAVING COUNT(*) > 10
ORDER BY avg_score ASC

#!/bin/bash
# NVLink健康检查脚本
for device in {0..7}; do
  bw=$(nvidia-smi topo -m | grep "GPU$device" | awk '{print $NF}')
  if [[ $bw != *"GB/s"* ]]; then
    echo "ALERT: Invalid bandwidth on GPU$device"
    systemctl restart nvidia-fabricmanager
    break
  fi
  
  speed=$(echo $bw | sed 's/GB\/s//')
  if (( $(echo "$speed < 40" | bc -l) )); then
    echo "WARNING: Low bandwidth ${speed}GB/s on GPU$device"
    # 自动迁移受影响Pod
    kubectl cordon node-$NODE
    kubectl drain node-$NODE --grace-period=300
  fi
done

apiVersion: v1
kind: ConfigMap
metadata:
  name: mig-policy
data:
  config: |
    {
      "mig-strategy": "mixed",
      "sharing": {
        "default": "none",
        "mig-devices": {
          "gpu": {
            "1g.5gb": { "replicas": 7 },
            "2g.10gb": { "replicas": 3 }
          }
        }
      },
      "topology-aware": true
    }

// 时间片调度核心逻辑
void schedule_time_slices(struct vgpu_scheduler *sched)
{
    while (!kthread_should_stop()) {
        for (i = 0; i < sched->num_vgpu; i++) {
            struct vgpu_instance *vgpu = &sched->vgpus[i];
            
            // 切换上下文
            if (current_vgpu != vgpu) {
                save_gpu_state(current_vgpu);
                load_gpu_state(vgpu);
                current_vgpu = vgpu;
            }
            
            // 执行时间片
            set_timer(sched->time_slice);
            wait_event_interruptible_timeout(
                sched->wait_queue, 
                timer_expired, 
                msecs_to_jiffies(sched->time_slice)
            );
        }
    }
}

def quantum_annealing_schedule(topology):
    # 构建QUBO模型
    qubo = build_qubo_model(topology)
    
    # 量子退火求解
    sampler = DWaveSampler()
    response = sampler.sample_qubo(qubo, num_reads=1000)
    
    # 提取最优分组
    best_solution = response.first.sample
    return decode_solution(best_solution)