登录/ 注册
 找回密码
 立即注册
关闭

CSDN热搜

程序园 精品问答 技术交流 资源下载
返回列表 发新帖
首页 › 业界区 › 业界 ›开源 vGPU 方案 HAMi 原理分析 Part1：hami-device-plug ...

开源 vGPU 方案 HAMi 原理分析 Part1：hami-device-plugin-nvidia 实现

系味 4 天前
1.png

本文为开源的 vGPU 方案 HAMi 实现原理分析第一篇，主要分析 hami-device-plugin-nvidia 实现原理。
之前在 开源 vGPU 方案：HAMi,实现细粒度 GPU 切分 介绍了 HAMi 是什么，然后在开源 vGPU 方案 HAMi: core&memory 隔离测试 中对 HAMi 提供的 vGPU 方案进行了测试。
接下来则是逐步分析 HAMi 中的 vGPU 实现原理，涉及到的东西比较多，暂定分为几部分：

  • 1）hami-device-plugin-nvidia：HAMi 版本的 device plugin 中 GPU 感知以及分配逻辑是怎么实现的，和 NVIDIA 原生的 device-plugin 有和不同。
  • 2）HAMI-Scheduler：HAMi 是如何做调度的，binpack/spread 高级调度策略是怎么实现的
  • 3）HAMi-Core：这也是 vCUDA 方案最核心的部分，HAMi 是如何通过拦截 CUDA API 实现 Core&Memory 隔离限制的
本文为第一篇，分析 hami-device-plugin-nvidia 实现原理。
1. 概述

NVIDIA 是有自己实现 device plugin 的，那么问题来了：HAMi 为什么还要自己实现一个 device plugin 呢？
是 hami-device-plugin-nvidia 是有哪些 NVIDIA 原生  device plugin 没有的功能吗？带着疑问，我们开始查看 hami-device-plugin-nvidia 源码。
这部分需要大家对 GPU Operator、k8s device plugin 等比较熟悉阅读起来才比较丝滑。
推荐阅读：

  • GPU 环境搭建指南：如何在裸机、Docker、K8s 等环境中使用 GPU
  • GPU 环境搭建指南：使用 GPU Operator 加速 Kubernetes GPU 环境搭建
  • Kubernetes教程(二一)—自定义资源支持：K8s Device Plugin 从原理到实现
  • Kubernetes教程(二二)—在 K8S 中创建 Pod 是如何使用到 GPU 的：device plugin&nvidia-container-toolkit 源码分析
后续都默认大家都对这块比较熟悉了，特别是后两篇
2. 程序入口

HAMi 首先支持的是 NVIDIA GPU，单独实现了一个 device plugin nvidia。

  • 启动文件在 cmd/device-plugin/nvidia
  • 核心实现在 pkg/device-plugin/nvidiadevice
默认大家都对 k8s 的 device plugin 机制比较熟悉了，因此这里只分析核心代码逻辑，不然篇幅就太长了。
对于一个 device plugin 我们一般关注以下 3 个地方：

  • Register：将插件注册到 Kubelet 的，参数 ResourceName 比较重要
  • ListAndWatch：device plugin 是怎么感知 GPU 并上报的
  • Allocate：device plugin 是如何将 GPU 分配给 Pod 的
启动命令在 /cmd/device-plugin/nvidia，用的是 github.com/urfave/cli/v2 构建的一个命令行工具。
  1. func main() {
  2.     var configFile string
  4.     c := cli.NewApp()
  5.     c.Name = "NVIDIA Device Plugin"
  6.     c.Usage = "NVIDIA device plugin for Kubernetes"
  7.     c.Version = info.GetVersionString()
  8.     c.Action = func(ctx *cli.Context) error {
  9.        return start(ctx, c.Flags)
  10.     }
  12.     c.Flags = []cli.Flag{
  13.        &cli.StringFlag{
  14.           Name:    "mig-strategy",
  15.           Value:   spec.MigStrategyNone,
  16.           Usage:   "the desired strategy for exposing MIG devices on GPUs that support it:\n\t\t[none | single | mixed]",
  17.           EnvVars: []string{"MIG_STRATEGY"},
  18.        },
  19.        &cli.BoolFlag{
  20.           Name:    "fail-on-init-error",
  21.           Value:   true,
  22.           Usage:   "fail the plugin if an error is encountered during initialization, otherwise block indefinitely",
  23.           EnvVars: []string{"FAIL_ON_INIT_ERROR"},
  24.        },
  25.        &cli.StringFlag{
  26.           Name:    "nvidia-driver-root",
  27.           Value:   "/",
  28.           Usage:   "the root path for the NVIDIA driver installation (typical values are '/' or '/run/nvidia/driver')",
  29.           EnvVars: []string{"NVIDIA_DRIVER_ROOT"},
  30.        },
  31.        &cli.BoolFlag{
  32.           Name:    "pass-device-specs",
  33.           Value:   false,
  34.           Usage:   "pass the list of DeviceSpecs to the kubelet on Allocate()",
  35.           EnvVars: []string{"PASS_DEVICE_SPECS"},
  36.        },
  37.        &cli.StringSliceFlag{
  38.           Name:    "device-list-strategy",
  39.           Value:   cli.NewStringSlice(string(spec.DeviceListStrategyEnvvar)),
  40.           Usage:   "the desired strategy for passing the device list to the underlying runtime:\n\t\t[envvar | volume-mounts | cdi-annotations]",
  41.           EnvVars: []string{"DEVICE_LIST_STRATEGY"},
  42.        },
  43.        &cli.StringFlag{
  44.           Name:    "device-id-strategy",
  45.           Value:   spec.DeviceIDStrategyUUID,
  46.           Usage:   "the desired strategy for passing device IDs to the underlying runtime:\n\t\t[uuid | index]",
  47.           EnvVars: []string{"DEVICE_ID_STRATEGY"},
  48.        },
  49.        &cli.BoolFlag{
  50.           Name:    "gds-enabled",
  51.           Usage:   "ensure that containers are started with NVIDIA_GDS=enabled",
  52.           EnvVars: []string{"GDS_ENABLED"},
  53.        },
  54.        &cli.BoolFlag{
  55.           Name:    "mofed-enabled",
  56.           Usage:   "ensure that containers are started with NVIDIA_MOFED=enabled",
  57.           EnvVars: []string{"MOFED_ENABLED"},
  58.        },
  59.        &cli.StringFlag{
  60.           Name:        "config-file",
  61.           Usage:       "the path to a config file as an alternative to command line options or environment variables",
  62.           Destination: &configFile,
  63.           EnvVars:     []string{"CONFIG_FILE"},
  64.        },
  65.        &cli.StringFlag{
  66.           Name:    "cdi-annotation-prefix",
  67.           Value:   spec.DefaultCDIAnnotationPrefix,
  68.           Usage:   "the prefix to use for CDI container annotation keys",
  69.           EnvVars: []string{"CDI_ANNOTATION_PREFIX"},
  70.        },
  71.        &cli.StringFlag{
  72.           Name:    "nvidia-ctk-path",
  73.           Value:   spec.DefaultNvidiaCTKPath,
  74.           Usage:   "the path to use for the nvidia-ctk in the generated CDI specification",
  75.           EnvVars: []string{"NVIDIA_CTK_PATH"},
  76.        },
  77.        &cli.StringFlag{
  78.           Name:    "container-driver-root",
  79.           Value:   spec.DefaultContainerDriverRoot,
  80.           Usage:   "the path where the NVIDIA driver root is mounted in the container; used for generating CDI specifications",
  81.           EnvVars: []string{"CONTAINER_DRIVER_ROOT"},
  82.        },
  83.     }
  84.     c.Flags = append(c.Flags, addFlags()...)
  85.     err := c.Run(os.Args)
  86.     if err != nil {
  87.        klog.Error(err)
  88.        os.Exit(1)
  89.     }
  90. }
  92. func addFlags() []cli.Flag {
  93.     addition := []cli.Flag{
  94.        &cli.StringFlag{
  95.           Name:    "node-name",
  96.           Value:   os.Getenv(util.NodeNameEnvName),
  97.           Usage:   "node name",
  98.           EnvVars: []string{"NodeName"},
  99.        },
  100.        &cli.UintFlag{
  101.           Name:    "device-split-count",
  102.           Value:   2,
  103.           Usage:   "the number for NVIDIA device split",
  104.           EnvVars: []string{"DEVICE_SPLIT_COUNT"},
  105.        },
  106.        &cli.Float64Flag{
  107.           Name:    "device-memory-scaling",
  108.           Value:   1.0,
  109.           Usage:   "the ratio for NVIDIA device memory scaling",
  110.           EnvVars: []string{"DEVICE_MEMORY_SCALING"},
  111.        },
  112.        &cli.Float64Flag{
  113.           Name:    "device-cores-scaling",
  114.           Value:   1.0,
  115.           Usage:   "the ratio for NVIDIA device cores scaling",
  116.           EnvVars: []string{"DEVICE_CORES_SCALING"},
  117.        },
  118.        &cli.BoolFlag{
  119.           Name:    "disable-core-limit",
  120.           Value:   false,
  121.           Usage:   "If set, the core utilization limit will be ignored",
  122.           EnvVars: []string{"DISABLE_CORE_LIMIT"},
  123.        },
  124.        &cli.StringFlag{
  125.           Name:  "resource-name",
  126.           Value: "nvidia.com/gpu",
  127.           Usage: "the name of field for number GPU visible in container",
  128.        },
  129.     }
  130.     return addition
  131. }
复制代码
启动时做了两件事：

  • 将插件注册到 Kubelet
  • 启动一个 gRPC 服务
我们只需要关注一下接收的几个参数：
  1. &cli.UintFlag{
  2.     Name:    "device-split-count",
  3.     Value:   2,
  4.     Usage:   "the number for NVIDIA device split",
  5.     EnvVars: []string{"DEVICE_SPLIT_COUNT"},
  6. },
  7. &cli.Float64Flag{
  8.     Name:    "device-memory-scaling",
  9.     Value:   1.0,
  10.     Usage:   "the ratio for NVIDIA device memory scaling",
  11.     EnvVars: []string{"DEVICE_MEMORY_SCALING"},
  12. },
  13. &cli.Float64Flag{
  14.     Name:    "device-cores-scaling",
  15.     Value:   1.0,
  16.     Usage:   "the ratio for NVIDIA device cores scaling",
  17.     EnvVars: []string{"DEVICE_CORES_SCALING"},
  18. },
  19. &cli.BoolFlag{
  20.     Name:    "disable-core-limit",
  21.     Value:   false,
  22.     Usage:   "If set, the core utilization limit will be ignored",
  23.     EnvVars: []string{"DISABLE_CORE_LIMIT"},
  24. },
  25. &cli.StringFlag{
  26.     Name:  "resource-name",
  27.     Value: "nvidia.com/gpu",
  28.     Usage: "the name of field for number GPU visible in container",
  29. },
复制代码

  • device-split-count：表示 GPU 的分割数，每一张 GPU 都不能分配超过其配置数目的任务。若其配置为 N 的话，每个 GPU 上最多可以同时存在 N 个任务。

    • 建议根据 GPU 性能动态调整，一般建议大于 10。

  • device-memory-scaling：表示 GPU memory 的 oversubscription(超额订阅)** **比例，默认 1.0，大于 1.0 则表示启用虚拟显存（实验功能），不建议修改。
  • device-cores-scaling：表示 GPU core 的 oversubscription(超额订阅)比例，默认 1.0。
  • disable-core-limit：是否关闭 GPU Core Limit，默认 false，不建议修改。
  • resource-name：资源名称，建议改掉，不推荐使用默认的 nvidia.com/gpu 因为这个和 nvidia 原生的重复了。
3. Register

Register

Register 方法实现如下：
  1. // pkg/device-plugin/nvidiadevice/nvinternal/plugin/server.go#L222
  3. // Register registers the device plugin for the given resourceName with Kubelet.
  4. func (plugin *NvidiaDevicePlugin) Register() error {
  5.     conn, err := plugin.dial(kubeletdevicepluginv1beta1.KubeletSocket, 5*time.Second)
  6.     if err != nil {
  7.        return err
  8.     }
  9.     defer conn.Close()
  11.     client := kubeletdevicepluginv1beta1.NewRegistrationClient(conn)
  12.     reqt := &kubeletdevicepluginv1beta1.RegisterRequest{
  13.        Version:      kubeletdevicepluginv1beta1.Version,
  14.        Endpoint:     path.Base(plugin.socket),
  15.        ResourceName: string(plugin.rm.Resource()),
  16.        Options: &kubeletdevicepluginv1beta1.DevicePluginOptions{
  17.           GetPreferredAllocationAvailable: false,
  18.        },
  19.     }
  21.     _, err = client.Register(context.Background(), reqt)
  22.     if err != nil {
  23.        return err
  24.     }
  25.     return nil
  26. }
复制代码
device plugin 注册时的几个核心信息：

  • ResourceName：资源名称，这个和创建 Pod 时申请 vGPU 的资源名匹配时就会使用到该 device plugin。

    • 也是可以在 deivce plugin 启动时配置的，一般叫做 --resource-name=nvidia.com/vgpu

  • Version：device plugin 的版本，这里是 v1beta1
  • Endpoint：device plugin 的访问地址，Kubelet 会通过这个 sock 和 device plugin 进行交互。

    • hami 用的格式为：/var/lib/kubelet/device-plugins/nvidia-xxx.sock，其中 xxx 是 从 ResourceName 中解析出来的，比如 nvidia.com/vgpu 那么这里的 xx 就是后面的 vgpu。

假设我们都使用默认值，ResourceName 为 nvidia.com/vgpu，Endpoint 为 /var/lib/kubelet/device-plugins/nvidia-vgpu.sock。

  • 后续 Pod Resource 中申请使用 nvidia.com/vgpu 资源时，就会由该 device plugin 来处理，实现资源分配，Kubelet 则是通过 var/lib/kubelet/device-plugins/nvidia-vgpu.sock 这个 sock 文件调用 device plugin API。
  • 反之，我们在 Pod Resource 中申请使用 nvidia.com/gpu 时，这个 ResourceName 和 hami 插件不匹配，因此不由 hami device plugin nvidia 处理，而是由 nvidia 自己的  device plugin 进行处理。
WatchAndRegister

这个是 HAMi device plugin 中的一个特殊逻辑，将 node 上的 GPU 信息以 annotations 的形式添加到 Node 对象上。
这里是直接和 kube-apiserver 进行通信，而不是使用的传统的 device plugin 上报流程。
后续 HAMi-Scheduler 在进行调度时就会用到这边上报的 annotations 作为调度依据的一部分，分析 HAMi-Scheduler 时在仔细分析。
  1. func (plugin *NvidiaDevicePlugin) WatchAndRegister() {
  2.     klog.Info("Starting WatchAndRegister")
  3.     errorSleepInterval := time.Second * 5
  4.     successSleepInterval := time.Second * 30
  5.     for {
  6.        err := plugin.RegistrInAnnotation()
  7.        if err != nil {
  8.           klog.Errorf("Failed to register annotation: %v", err)
  9.           klog.Infof("Retrying in %v seconds...", errorSleepInterval)
  10.           time.Sleep(errorSleepInterval)
  11.        } else {
  12.           klog.Infof("Successfully registered annotation. Next check in %v seconds...", successSleepInterval)
  13.           time.Sleep(successSleepInterval)
  14.        }
  15.     }
  16. }
复制代码
getAPIDevices

获取 Node 上的 GPU 信息，并组装成 api.DeviceInfo 对象。
  1. func (plugin *NvidiaDevicePlugin) getAPIDevices() *[]*api.DeviceInfo {
  2.     devs := plugin.Devices()
  3.     nvml.Init()
  4.     res := make([]*api.DeviceInfo, 0, len(devs))
  5.     idx := 0
  6.     for idx < len(devs) {
  7.        ndev, ret := nvml.DeviceGetHandleByIndex(idx)
  8.        //ndev, err := nvml.NewDevice(uint(idx))
  9.        //klog.V(3).Infoln("ndev type=", ndev.Model)
  10.        if ret != nvml.SUCCESS {
  11.           klog.Errorln("nvml new device by index error idx=", idx, "err=", ret)
  12.           panic(0)
  13.        }
  14.        memoryTotal := 0
  15.        memory, ret := ndev.GetMemoryInfo()
  16.        if ret == nvml.SUCCESS {
  17.           memoryTotal = int(memory.Total)
  18.        } else {
  19.           klog.Error("nvml get memory error ret=", ret)
  20.           panic(0)
  21.        }
  22.        UUID, ret := ndev.GetUUID()
  23.        if ret != nvml.SUCCESS {
  24.           klog.Error("nvml get uuid error ret=", ret)
  25.           panic(0)
  26.        }
  27.        Model, ret := ndev.GetName()
  28.        if ret != nvml.SUCCESS {
  29.           klog.Error("nvml get name error ret=", ret)
  30.           panic(0)
  31.        }
  33.        registeredmem := int32(memoryTotal / 1024 / 1024)
  34.        if *util.DeviceMemoryScaling != 1 {
  35.           registeredmem = int32(float64(registeredmem) * *util.DeviceMemoryScaling)
  36.        }
  37.        klog.Infoln("MemoryScaling=", *util.DeviceMemoryScaling, "registeredmem=", registeredmem)
  38.        health := true
  39.        for _, val := range devs {
  40.           if strings.Compare(val.ID, UUID) == 0 {
  41.              // when NVIDIA-Tesla P4, the device info is : ID:GPU-e290caca-2f0c-9582-acab-67a142b61ffa,Health:Healthy,Topology:nil,
  42.              // it is more reasonable to think of healthy as case-insensitive
  43.              if strings.EqualFold(val.Health, "healthy") {
  44.                 health = true
  45.              } else {
  46.                 health = false
  47.              }
  48.              break
  49.           }
  50.        }
  51.        numa, err := plugin.getNumaInformation(idx)
  52.        if err != nil {
  53.           klog.ErrorS(err, "failed to get numa information", "idx", idx)
  54.        }
  55.        res = append(res, &api.DeviceInfo{
  56.           ID:      UUID,
  57.           Count:   int32(*util.DeviceSplitCount),
  58.           Devmem:  registeredmem,
  59.           Devcore: int32(*util.DeviceCoresScaling * 100),
  60.           Type:    fmt.Sprintf("%v-%v", "NVIDIA", Model),
  61.           Numa:    numa,
  62.           Health:  health,
  63.        })
  64.        idx++
  65.        klog.Infof("nvml registered device id=%v, memory=%v, type=%v, numa=%v", idx, registeredmem, Model, numa)
  66.     }
  67.     return &res
  68. }
复制代码
核心部分
  1. // 通过 nvml 库获取 GPU 信息
  2. ndev, ret := nvml.DeviceGetHandleByIndex(idx)
  3. memoryTotal := 0
  4. memory, ret := ndev.GetMemoryInfo()
  5. if ret == nvml.SUCCESS {
  6.     memoryTotal = int(memory.Total)
  7. }
  8. UUID, ret := ndev.GetUUID()
  9. Model, ret := ndev.GetName()
  11. // 处理 Scaling
  12. registeredmem := int32(memoryTotal / 1024 / 1024)
  13. if *util.DeviceMemoryScaling != 1 {
  14.     registeredmem = int32(float64(registeredmem) * *util.DeviceMemoryScaling)
  15. }
  16. // 组装结果返回
  17. res = append(res, &api.DeviceInfo{
  18.     ID:      UUID,
  19.     Count:   int32(*util.DeviceSplitCount),
  20.     Devmem:  registeredmem,
  21.     Devcore: int32(*util.DeviceCoresScaling * 100),
  22.     Type:    fmt.Sprintf("%v-%v", "NVIDIA", Model),
  23.     Numa:    numa,
  24.     Health:  health,
  25. })
复制代码
更新到 Node Annoations

拿到 Device 信息之后，调用 kube-apiserver 更新 Node 对象的 Annoations 把 Device 信息存起来。
  1. encodeddevices := util.EncodeNodeDevices(*devices)
  2. annos[nvidia.HandshakeAnnos] = "Reported " + time.Now().String()
  3. annos[nvidia.RegisterAnnos] = encodeddevices
  4. klog.Infof("patch node with the following annos %v", fmt.Sprintf("%v", annos))
  5. err = util.PatchNodeAnnotations(node, annos)
复制代码
正常应该是走 k8s 的 device plugin 接口上报信息才对，这里是 HAMi 的特殊逻辑。
Demo

查看 Node 上的 Annoations，看看这边记录了些什么数据
  1. root@j99cloudvm:~# k get node j99cloudvm -oyaml
  2. apiVersion: v1
  3. kind: Node
  4. metadata:
  5.   annotations:
  6.     hami.io/node-handshake: Requesting_2024.09.25 07:48:26
  7.     hami.io/node-nvidia-register: 'GPU-03f69c50-207a-2038-9b45-23cac89cb67d,10,46068,100,NVIDIA-NVIDIA
  8.       A40,0,true:GPU-1afede84-4e70-2174-49af-f07ebb94d1ae,10,46068,100,NVIDIA-NVIDIA
  9.       A40,0,true:'
复制代码
hami.io/node-nvidia-register 就是 HAMi 的 device plugin 更新到 Node 上的 GPU 信息，格式化一下
  1. GPU-03f69c50-207a-2038-9b45-23cac89cb67d,10,46068,100,NVIDIA-NVIDIA A40,0,true:
  2. GPU-1afede84-4e70-2174-49af-f07ebb94d1ae,10,46068,100,NVIDIA-NVIDIA A40,0,true:
复制代码
当前节点上是两张 A40 GPU，

  • GPU-03f69c50-207a-2038-9b45-23cac89cb67d：为 GPU 设备的 UUID
  • 10,46068,100： 切分为 10 份，每张卡 46068M 内存，core 为 100 个(说明没有配置 oversubscription)
  • NVIDIA-NVIDIA：GPU 类型
  • A40：GPU 型号
  • 0：表示 GPU 的 NUMA 结构
  • true：表示该 GPU 是健康的
  • : 最后的冒号是分隔符
ps：这部分信息后续 hami-scheduler 进行调度时会用到，这里暂时不管。
小结

Register 方法分为两部分：

  • Register：将 device plugin 注册到 kubelet
  • WatchAndRegister：感知 Node 上的 GPU 信息，并和 kube-apiserver 交互，将这部分信息以 annotations 的形式添加到 Node 对象上,以便后续 hami-scheduler 使用。
4. ListAndWatch

ListAndWatch 方法用于感知节点上的设备并上报给 Kubelet。
由于需要将同一个 GPU 切分给多个 Pod 使用，因此 HAMi 的 device plugin 也会有类似 TimeSlicing 中的 Device 复制操作。
具体实现如下：
  1. // ListAndWatch lists devices and update that list according to the health status
  2. func (plugin *NvidiaDevicePlugin) ListAndWatch(e *kubeletdevicepluginv1beta1.Empty, s kubeletdevicepluginv1beta1.DevicePlugin_ListAndWatchServer) error {
  3.     s.Send(&kubeletdevicepluginv1beta1.ListAndWatchResponse{Devices: plugin.apiDevices()})
  5.     for {
  6.        select {
  7.        case <-plugin.stop:
  8.           return nil
  9.        case d := <-plugin.health:
  10.           // FIXME: there is no way to recover from the Unhealthy state.
  11.           d.Health = kubeletdevicepluginv1beta1.Unhealthy
  12.           klog.Infof("'%s' device marked unhealthy: %s", plugin.rm.Resource(), d.ID)
  13.           s.Send(&kubeletdevicepluginv1beta1.ListAndWatchResponse{Devices: plugin.apiDevices()})
  14.        }
  15.     }
  16. }
复制代码
比较长，我们只需要关注核心部分，同时先忽略 MIG 相关的逻辑。
首先是添加一个 CUDA_DEVICE_MEMORY_LIMIT_$Index 的环境变量，用于 gpu memory 限制。
  1. // VisitDevices visits each top-level device and invokes a callback function for it
  2. func (d *devicelib) VisitDevices(visit func(int, Device) error) error {
  3.     count, ret := d.nvml.DeviceGetCount()
  4.     if ret != nvml.SUCCESS {
  5.        return fmt.Errorf("error getting device count: %v", ret)
  6.     }
  8.     for i := 0; i < count; i++ {
  9.        device, ret := d.nvml.DeviceGetHandleByIndex(i)
  10.        if ret != nvml.SUCCESS {
  11.           return fmt.Errorf("error getting device handle for index '%v': %v", i, ret)
  12.        }
  13.        dev, err := d.newDevice(device)
  14.        if err != nil {
  15.           return fmt.Errorf("error creating new device wrapper: %v", err)
  16.        }
  18.        isSkipped, err := dev.isSkipped()
  19.        if err != nil {
  20.           return fmt.Errorf("error checking whether device is skipped: %v", err)
  21.        }
  22.        if isSkipped {
  23.           continue
  24.        }
  26.        err = visit(i, dev)
  27.        if err != nil {
  28.           return fmt.Errorf("error visiting device: %v", err)
  29.        }
  30.     }
  31.     return nil
  32. }
  34. // buildGPUDeviceMap builds a map of resource names to GPU devices
  35. func (b *deviceMapBuilder) buildGPUDeviceMap() (DeviceMap, error) {
  36.     devices := make(DeviceMap)
  38.     b.VisitDevices(func(i int, gpu device.Device) error {
  39.        name, ret := gpu.GetName()
  40.        if ret != nvml.SUCCESS {
  41.           return fmt.Errorf("error getting product name for GPU: %v", ret)
  42.        }
  43.        migEnabled, err := gpu.IsMigEnabled()
  44.        if err != nil {
  45.           return fmt.Errorf("error checking if MIG is enabled on GPU: %v", err)
  46.        }
  47.        if migEnabled && *b.config.Flags.MigStrategy != spec.MigStrategyNone {
  48.           return nil
  49.        }
  50.        for _, resource := range b.config.Resources.GPUs {
  51.           if resource.Pattern.Matches(name) {
  52.              index, info := newGPUDevice(i, gpu)
  53.              return devices.setEntry(resource.Name, index, info)
  54.           }
  55.        }
  56.        return fmt.Errorf("GPU name '%v' does not match any resource patterns", name)
  57.     })
  58.     return devices, nil
  59. }
复制代码
然后则是根据申请的 gpucores 配置 gpu core 限制的环境变量
  1. // GetPluginDevices returns the plugin Devices from all devices in the Devices
  2. func (ds Devices) GetPluginDevices() []*kubeletdevicepluginv1beta1.Device {
  3.     var res []*kubeletdevicepluginv1beta1.Device
  5.     if !strings.Contains(ds.GetIDs()[0], "MIG") {
  6.        for _, dev := range ds {
  7.           for i := uint(0); i < *util.DeviceSplitCount; i++ {
  8.              id := fmt.Sprintf("%v-%v", dev.ID, i)
  9.              res = append(res, &kubeletdevicepluginv1beta1.Device{
  10.                 ID:       id,
  11.                 Health:   dev.Health,
  12.                 Topology: nil,
  13.              })
  14.           }
  15.        }
  16.     } else {
  17.        for _, d := range ds {
  18.           res = append(res, &d.Device)
  19.        }
  21.     }
  23.     return res
  24. }
复制代码
这个用于设置 share_region  mmap 文件在容器中的位置
  1. for _, dev := range ds {
  2.   for i := uint(0); i < *util.DeviceSplitCount; i++ {
  3.      id := fmt.Sprintf("%v-%v", dev.ID, i)
  4.      res = append(res, &kubeletdevicepluginv1beta1.Device{
  5.         ID:       id,
  6.         Health:   dev.Health,
  7.         Topology: nil,
  8.      })
  9.   }
复制代码
Gpu memory 超额订阅
  1. // pkg/device-plugin/nvidiadevice/nvinternal/plugin/server.go#L290
  2. func (plugin *NvidiaDevicePlugin) Allocate(ctx context.Context, reqs *kubeletdevicepluginv1beta1.AllocateRequest) (*kubeletdevicepluginv1beta1.AllocateResponse, error) {
  3.     klog.InfoS("Allocate", "request", reqs)
  4.     responses := kubeletdevicepluginv1beta1.AllocateResponse{}
  5.     nodename := os.Getenv(util.NodeNameEnvName)
  6.     current, err := util.GetPendingPod(ctx, nodename)
  7.     if err != nil {
  8.        nodelock.ReleaseNodeLock(nodename, NodeLockNvidia)
  9.        return &kubeletdevicepluginv1beta1.AllocateResponse{}, err
  10.     }
  11.     klog.V(5).Infof("allocate pod name is %s/%s, annotation is %+v", current.Namespace, current.Name, current.Annotations)
  13.     for idx, req := range reqs.ContainerRequests {
  14.        // If the devices being allocated are replicas, then (conditionally)
  15.        // error out if more than one resource is being allocated.
  17.        if strings.Contains(req.DevicesIDs[0], "MIG") {
  19.           if plugin.config.Sharing.TimeSlicing.FailRequestsGreaterThanOne && rm.AnnotatedIDs(req.DevicesIDs).AnyHasAnnotations() {
  20.              if len(req.DevicesIDs) > 1 {
  21.                 return nil, fmt.Errorf("request for '%v: %v' too large: maximum request size for shared resources is 1", plugin.rm.Resource(), len(req.DevicesIDs))
  22.              }
  23.           }
  25.           for _, id := range req.DevicesIDs {
  26.              if !plugin.rm.Devices().Contains(id) {
  27.                 return nil, fmt.Errorf("invalid allocation request for '%s': unknown device: %s", plugin.rm.Resource(), id)
  28.              }
  29.           }
  31.           response, err := plugin.getAllocateResponse(req.DevicesIDs)
  32.           if err != nil {
  33.              return nil, fmt.Errorf("failed to get allocate response: %v", err)
  34.           }
  35.           responses.ContainerResponses = append(responses.ContainerResponses, response)
  36.        } else {
  37.           currentCtr, devreq, err := util.GetNextDeviceRequest(nvidia.NvidiaGPUDevice, *current)
  38.           klog.Infoln("deviceAllocateFromAnnotation=", devreq)
  39.           if err != nil {
  40.              device.PodAllocationFailed(nodename, current, NodeLockNvidia)
  41.              return &kubeletdevicepluginv1beta1.AllocateResponse{}, err
  42.           }
  43.           if len(devreq) != len(reqs.ContainerRequests[idx].DevicesIDs) {
  44.              device.PodAllocationFailed(nodename, current, NodeLockNvidia)
  45.              return &kubeletdevicepluginv1beta1.AllocateResponse{}, errors.New("device number not matched")
  46.           }
  47.           response, err := plugin.getAllocateResponse(util.GetContainerDeviceStrArray(devreq))
  48.           if err != nil {
  49.              return nil, fmt.Errorf("failed to get allocate response: %v", err)
  50.           }
  52.           err = util.EraseNextDeviceTypeFromAnnotation(nvidia.NvidiaGPUDevice, *current)
  53.           if err != nil {
  54.              device.PodAllocationFailed(nodename, current, NodeLockNvidia)
  55.              return &kubeletdevicepluginv1beta1.AllocateResponse{}, err
  56.           }
  58.           for i, dev := range devreq {
  59.              limitKey := fmt.Sprintf("CUDA_DEVICE_MEMORY_LIMIT_%v", i)
  60.              response.Envs[limitKey] = fmt.Sprintf("%vm", dev.Usedmem)
  62.              /*tmp := response.Envs["NVIDIA_VISIBLE_DEVICES"]
  63.              if i > 0 {
  64.                 response.Envs["NVIDIA_VISIBLE_DEVICES"] = fmt.Sprintf("%v,%v", tmp, dev.UUID)
  65.              } else {
  66.                 response.Envs["NVIDIA_VISIBLE_DEVICES"] = dev.UUID
  67.              }*/
  68.           }
  69.           response.Envs["CUDA_DEVICE_SM_LIMIT"] = fmt.Sprint(devreq[0].Usedcores)
  70.           response.Envs["CUDA_DEVICE_MEMORY_SHARED_CACHE"] = fmt.Sprintf("%s/vgpu/%v.cache", hostHookPath, uuid.New().String())
  71.           if *util.DeviceMemoryScaling > 1 {
  72.              response.Envs["CUDA_OVERSUBSCRIBE"] = "true"
  73.           }
  74.           if *util.DisableCoreLimit {
  75.              response.Envs[api.CoreLimitSwitch] = "disable"
  76.           }
  77.           cacheFileHostDirectory := fmt.Sprintf("%s/vgpu/containers/%s_%s", hostHookPath, current.UID, currentCtr.Name)
  78.           os.RemoveAll(cacheFileHostDirectory)
  80.           os.MkdirAll(cacheFileHostDirectory, 0777)
  81.           os.Chmod(cacheFileHostDirectory, 0777)
  82.           os.MkdirAll("/tmp/vgpulock", 0777)
  83.           os.Chmod("/tmp/vgpulock", 0777)
  84.           response.Mounts = append(response.Mounts,
  85.              &kubeletdevicepluginv1beta1.Mount{ContainerPath: fmt.Sprintf("%s/vgpu/libvgpu.so", hostHookPath),
  86.                 HostPath: hostHookPath + "/vgpu/libvgpu.so",
  87.                 ReadOnly: true},
  88.              &kubeletdevicepluginv1beta1.Mount{ContainerPath: fmt.Sprintf("%s/vgpu", hostHookPath),
  89.                 HostPath: cacheFileHostDirectory,
  90.                 ReadOnly: false},
  91.              &kubeletdevicepluginv1beta1.Mount{ContainerPath: "/tmp/vgpulock",
  92.                 HostPath: "/tmp/vgpulock",
  93.                 ReadOnly: false},
  94.           )
  95.           found := false
  96.           for _, val := range currentCtr.Env {
  97.              if strings.Compare(val.Name, "CUDA_DISABLE_CONTROL") == 0 {
  98.                 // if env existed but is set to false or can not be parsed, ignore
  99.                 t, _ := strconv.ParseBool(val.Value)
  100.                 if !t {
  101.                    continue
  102.                 }
  103.                 // only env existed and set to true, we mark it "found"
  104.                 found = true
  105.                 break
  106.              }
  107.           }
  108.           if !found {
  109.              response.Mounts = append(response.Mounts, &kubeletdevicepluginv1beta1.Mount{ContainerPath: "/etc/ld.so.preload",
  110.                 HostPath: hostHookPath + "/vgpu/ld.so.preload",
  111.                 ReadOnly: true},
  112.              )
  113.           }
  114.           _, err = os.Stat(fmt.Sprintf("%s/vgpu/license", hostHookPath))
  115.           if err == nil {
  116.              response.Mounts = append(response.Mounts, &kubeletdevicepluginv1beta1.Mount{
  117.                 ContainerPath: "/tmp/license",
  118.                 HostPath:      fmt.Sprintf("%s/vgpu/license", hostHookPath),
  119.                 ReadOnly:      true,
  120.              })
  121.              response.Mounts = append(response.Mounts, &kubeletdevicepluginv1beta1.Mount{
  122.                 ContainerPath: "/usr/bin/vgpuvalidator",
  123.                 HostPath:      fmt.Sprintf("%s/vgpu/vgpuvalidator", hostHookPath),
  124.                 ReadOnly:      true,
  125.              })
  126.           }
  127.           responses.ContainerResponses = append(responses.ContainerResponses, response)
  128.        }
  129.     }
  130.     klog.Infoln("Allocate Response", responses.ContainerResponses)
  131.     device.PodAllocationTrySuccess(nodename, nvidia.NvidiaGPUDevice, NodeLockNvidia, current)
  132.     return &responses, nil
  133. }
复制代码
是否关闭算力限制
  1. for i, dev := range devreq {
  2.     limitKey := fmt.Sprintf("CUDA_DEVICE_MEMORY_LIMIT_%v", i)
  3.     response.Envs[limitKey] = fmt.Sprintf("%vm", dev.Usedmem)
  5.     /*tmp := response.Envs["NVIDIA_VISIBLE_DEVICES"]
  6.     if i > 0 {
  7.        response.Envs["NVIDIA_VISIBLE_DEVICES"] = fmt.Sprintf("%v,%v", tmp, dev.UUID)
  8.     } else {
  9.        response.Envs["NVIDIA_VISIBLE_DEVICES"] = dev.UUID
  10.     }*/
  11. }
复制代码
挂载 vgpu 相关文件
这里就实现了 libvgpu.so 库的替换。
  1. response.Envs["CUDA_DEVICE_SM_LIMIT"] = fmt.Sprint(devreq[0].Usedcores)
复制代码
替换动态库,当没有指定  CUDA_DISABLE_CONTROL=true 时就会做该处理
  1. response.Envs["CUDA_DEVICE_MEMORY_SHARED_CACHE"] = fmt.Sprintf("%s/vgpu/%v.cache", hostHookPath, uuid.New().String())
复制代码
整个实现也算比较容易理解，就是给 Pod 增加了一系列环境变量，以及增加了替换 libvgpu.so 的 Mounts 配置，后续这个 libvgpu.so 就会根据这些环境变量做 Core&Memory 的限制。
NVIDIA 原生逻辑
  1. if *util.DeviceMemoryScaling > 1 {
  2.     response.Envs["CUDA_OVERSUBSCRIBE"] = "true"
  3. }
复制代码
核心部分就是这一句，添加了一个环境变量
  1. if *util.DisableCoreLimit {
  2.     response.Envs[api.CoreLimitSwitch] = "disable"
  3. }
复制代码
而 plugin.deviceListEnvvar 的值来自：
  1. // 缓存文件存放位置 /usr/local/vgpu/containers/xxx/xxx
  2. cacheFileHostDirectory := fmt.Sprintf("%s/vgpu/containers/%s_%s", hostHookPath, current.UID, currentCtr.Name)
  3. os.RemoveAll(cacheFileHostDirectory)
  5. os.MkdirAll(cacheFileHostDirectory, 0777)
  6. os.Chmod(cacheFileHostDirectory, 0777)
  7. os.MkdirAll("/tmp/vgpulock", 0777)
  8. os.Chmod("/tmp/vgpulock", 0777)
  10. response.Mounts = append(response.Mounts,
  11.     // 宿主机上的 libvgpu.so挂载到 pod 里替换 nvidia 默认的驱动
  12.     &kubeletdevicepluginv1beta1.Mount{ContainerPath: fmt.Sprintf("%s/vgpu/libvgpu.so", hostHookPath),
  13.        HostPath: hostHookPath + "/vgpu/libvgpu.so",
  14.        ReadOnly: true},
  15.     // 随机的文件挂载进 pod 作为 vgpu 使用
  16.     &kubeletdevicepluginv1beta1.Mount{ContainerPath: fmt.Sprintf("%s/vgpu", hostHookPath),
  17.        HostPath: cacheFileHostDirectory,
  18.        ReadOnly: false},
  19.     // 一个 lock 文件
  20.     &kubeletdevicepluginv1beta1.Mount{ContainerPath: "/tmp/vgpulock",
  21.        HostPath: "/tmp/vgpulock",
  22.        ReadOnly: false},
  23. )
复制代码
即：NVIDIA_VISIBLE_DEVICES
正好，这个 ENV 就是 NVIDIA  deviceplugin 中的实现，设置该环境变量之后 nvidia-container-toolkit 会为有这个环境变量的容器分配 GPU。
HAMi  这里则是复用了  nvidia-container-toolkit  的能力将 GPU 分配给 Pod。
小结

Allocate 方法中核心部分包括三件事情：

  • HAMi 自定义逻辑

    • 添加用于资源限制的环境变量CUDA_DEVICE_MEMORY_LIMIT_X 和 CUDA_DEVICE_SM_LIMIT
    • 挂载 libvgpu.so 到 Pod 中进行替换

  • NVIDIA 原生逻辑

    • 添加用于分配 GPU 的 环境变量 NVIDIA_VISIBLE_DEVICES,借助 NVIDIA Container Toolkit 将 GPU 挂载到 Pod 里

6. 总结

至此，HAMi 的 NVIDIA device plugin 工作原理就很清晰了。

  • 首先是 Register 插件注册，可以配置使用和原生 nvidia device plugin 不同的 ResourceName 来进行区分。

    • 另外会额外启动一个后台 goroutine WatchAndRegister 定时将 GPU 信息更新到 Node 对象的 Annoations 上便于 Scheduler 时使用。

  • 然后 ListAndWatch 感知设备时也根据配置对 Device 进行复制，让同一个设备能够分配给多个 Pod。

    • 这个和 TimeSlicing 方案一样

  • 最后 Allocate 方法中主要做三件事：

    • 1）为容器中增加NVIDIA_VISIBLE_DEVICES 环境变量，借助 NVIDIA Container Toolkit 实现为容器分配 GPU
    • 2）增加 Mounts 配置，挂载 libvgpu.so 到容器中实现对原始驱动的替换
    • 3）为容器中增加部分 HAMi 自定义的环境变量 CUDA_DEVICE_MEMORY_LIMIT_X 和 CUDA_DEVICE_SM_LIMIT ，配合 ibvgpu.so 实现 GPU core、memory 的限制

核心其实就是在 Allocate 方法中，给容器中添加CUDA_DEVICE_MEMORY_LIMIT_X 和 CUDA_DEVICE_SM_LIMIT环境变量和挂载 libvgpu.so 到容器中实现对原始驱动的替换。
当容器启动后，CUDA API 请求先经过 libvgpu.so，然后 libvgpu.so 根据环境变量 CUDA_DEVICE_MEMORY_LIMIT_X 和 CUDA_DEVICE_SM_LIMIT 实现 Core & Memory 限制。
最后回答开篇提出的问题：HAMi 为什么要自己实现一个 device plugin 呢？hami-device-plugin-nvidia 是有哪些 NVIDIA 原生  device plugin 没有的功能吗？
在 hami device plugin 相比原生的 NVIDIA device plugin 做了几个修改：

  • 1）注册时额外启动后台 goroutine WatchAndRegister 定时将 GPU 信息更新到 Node 对象的 Annoations 上便于 Scheduler 时使用。
  • 2）ListAndWatch 感知设备时也根据配置对 Device 进行复制，便于将同一个物理 GPU 分配给多个 Pod

    • 这个其实原生的 NVIDIA device plugin 也有，就是 TimeSlicing 方案。

  • 3）Allocate 中增加了 HAMi 自定义逻辑：

    • 挂载 libvgpu.so 到容器中实现对原始驱动的替换
    • 指定部分 HAMi 自定义的环境变量CUDA_DEVICE_MEMORY_LIMIT_X 和 CUDA_DEVICE_SM_LIMIT，配合 ibvgpu.so 实现 GPU core、memory 的限制

7. FAQ

Node 上的 libvgpu.so 是怎么来的

Allocate 方法中要将 libvgpu.so 挂载到 Pod 里，这里用的是 HostPath 方式挂载，说明这个 libvgpu.so 是存在于宿主机上的。
那么问题来了，宿主机上的 libvgpu.so 是怎么来的?
这个实际上是打包在 HAMi 提供的 device-plugin 镜像里的，device-plugin 启动时将其从 Pod 里复制到宿主机上，相关 yaml 如下：
  1. found := false
  2. for _, val := range currentCtr.Env {
  3.     if strings.Compare(val.Name, "CUDA_DISABLE_CONTROL") == 0 {
  4.        // if env existed but is set to false or can not be parsed, ignore
  5.        t, _ := strconv.ParseBool(val.Value)
  6.        if !t {
  7.           continue
  8.        }
  9.        // only env existed and set to true, we mark it "found"
  10.        found = true
  11.        break
  12.     }
  13. }
  14. if !found {
  15.     response.Mounts = append(response.Mounts, &kubeletdevicepluginv1beta1.Mount{ContainerPath: "/etc/ld.so.preload",
  16.        HostPath: hostHookPath + "/vgpu/ld.so.preload",
  17.        ReadOnly: true},
  18.     )
  19. }
复制代码
挂载了 hostPath 到容器里，然后容器里执行cp -f /k8s-vgpu/lib/nvidia/* /usr/local/vgpu/ 命令将其复制到宿主机。
【Kubernetes 系列】持续更新中，搜索公众号【探索云原生】订阅，阅读更多文章。
2.png


来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！
开源vGPU方案HAMi原理

使用道具 举报

相关推荐
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册
发帖
系味
4 天前

0

粉丝关注

11

主题发布

板块介绍填写区域，请于后台编辑
微信扫一扫