目录

  • 每日一句正能量
  • 一、引言：为什么多任务调度如此重要？
  • 二、任务调度基础
  • 2.1 调度算法
  • 2.2 任务调度实现
  • 三、资源分配
  • 3.1 资源分配策略
  • 3.2 资源分配实现
  • 四、并发控制
  • 4.1 并发控制机制
  • 4.2 并发控制实现
  • 五、性能监控
  • 5.1 性能监控指标
  • 5.2 性能监控实现
  • 六、实战案例
  • 6.1 多任务AI应用
  • 七、常见问题与解决
  • 八、结语：多任务调度是端侧AI的"大脑"

在这里插入图片描述

每日一句正能量

“前半生折腾，后半生沉淀。”
人生不同阶段的不同价值与使命。尽情地活过，然后深刻地懂得。前半生的所有“折腾”，都是后半生得以“沉淀”的宝贵材料。

相信"多任务调度是端侧AI的大脑"。

摘要：多任务调度是端侧AI的核心挑战。本文深入探讨端侧AI的多任务调度策略，从优先级队列、资源分配到并发控制，提供HarmonyOS多任务调度的完整实现方案，帮助开发者打造高效的端侧AI调度系统。

一、引言：为什么多任务调度如此重要？

"多个AI任务同时运行，系统卡顿严重。"

"高优先级任务被低优先级任务阻塞，用户体验差。"

"资源分配不合理，导致部分任务无法完成。"

端侧AI面临多任务并发的挑战：

  • 任务多样性：图像识别、语音识别、自然语言处理等多种任务
  • 资源有限性：CPU、NPU、内存等资源有限
  • 实时性要求：部分任务需要实时响应
  • 功耗约束：需要在有限功耗下完成任务

二、任务调度基础

2.1 调度算法

在这里插入图片描述

图1：调度算法——FCFS、SJF、优先级、RR

算法描述优点缺点适用场景
FCFS先来先服务简单公平平均等待时间长批处理
SJF短作业优先平均等待时间短可能饥饿实时系统
优先级按优先级调度重要任务优先低优先级饥饿实时系统
RR时间片轮转响应时间公平上下文开销大交互系统
EDF最早截止时间优先满足截止时间实现复杂实时系统

2.2 任务调度实现

// 任务调度器
class TaskScheduler {
  private taskQueue: AITask[] = []
  private runningTasks: Map<string, AITask> = new Map()
  private maxConcurrentTasks: number = 4
  private scheduler: SchedulerAlgorithm

  constructor(algorithm: SchedulerAlgorithm = 'priority') {
    this.scheduler = algorithm
  }

  // 添加任务
  addTask(task: AITask): void {
    this.taskQueue.push(task)
    this.sortQueue()
  }

  // 执行任务调度
  async schedule(): Promise<void> {
    while (this.taskQueue.length > 0 || this.runningTasks.size > 0) {
      // 检查是否有空闲资源
      if (this.runningTasks.size < this.maxConcurrentTasks) {
        const task = this.selectNextTask()
        if (task) {
          this.startTask(task)
        }
      }

      // 等待一段时间
      await this.delay(10)
    }
  }

  // 选择下一个任务
  private selectNextTask(): AITask | null {
    if (this.taskQueue.length === 0) {
      return null
    }

    switch (this.scheduler) {
      case 'fcfs':
        return this.selectFCFS()
      case 'sjf':
        return this.selectSJF()
      case 'priority':
        return this.selectPriority()
      case 'rr':
        return this.selectRR()
      case 'edf':
        return this.selectEDF()
      default:
        return this.selectFCFS()
    }
  }

  // FCFS选择
  private selectFCFS(): AITask | null {
    return this.taskQueue.shift() || null
  }

  // SJF选择
  private selectSJF(): AITask | null {
    if (this.taskQueue.length === 0) return null

    let shortestIndex = 0
    for (let i = 1; i < this.taskQueue.length; i++) {
      if (this.taskQueue[i].estimatedTime < this.taskQueue[shortestIndex].estimatedTime) {
        shortestIndex = i
      }
    }

    return this.taskQueue.splice(shortestIndex, 1)[0]
  }

  // 优先级选择
  private selectPriority(): AITask | null {
    if (this.taskQueue.length === 0) return null

    let highestIndex = 0
    for (let i = 1; i < this.taskQueue.length; i++) {
      if (this.taskQueue[i].priority > this.taskQueue[highestIndex].priority) {
        highestIndex = i
      }
    }

    return this.taskQueue.splice(highestIndex, 1)[0]
  }

  // RR选择
  private selectRR(): AITask | null {
    return this.taskQueue.shift() || null
  }

  // EDF选择
  private selectEDF(): AITask | null {
    if (this.taskQueue.length === 0) return null

    let earliestIndex = 0
    for (let i = 1; i < this.taskQueue.length; i++) {
      if (this.taskQueue[i].deadline < this.taskQueue[earliestIndex].deadline) {
        earliestIndex = i
      }
    }

    return this.taskQueue.splice(earliestIndex, 1)[0]
  }

  // 开始任务
  private startTask(task: AITask): void {
    this.runningTasks.set(task.id, task)
    
    // 执行任务
    this.executeTask(task).then(() => {
      this.runningTasks.delete(task.id)
    })
  }

  // 执行任务
  private async executeTask(task: AITask): Promise<void> {
    // 执行任务的实现
    console.log(`Executing task: ${task.name}`)
    await this.delay(task.estimatedTime)
  }

  // 排序队列
  private sortQueue(): void {
    switch (this.scheduler) {
      case 'priority':
        this.taskQueue.sort((a, b) => b.priority - a.priority)
        break
      case 'edf':
        this.taskQueue.sort((a, b) => a.deadline - b.deadline)
        break
      default:
        break
    }
  }

  // 延迟
  private delay(ms: number): Promise<void> {
    return new Promise(resolve => setTimeout(resolve, ms))
  }
}

// AI任务
interface AITask {
  id: string
  name: string
  priority: number
  estimatedTime: number
  deadline: number
  resourceRequirements: ResourceRequirements
}

// 资源需求
interface ResourceRequirements {
  cpu: number
  memory: number
  npu: number
}

// 调度算法
enum SchedulerAlgorithm {
  FCFS = 'fcfs',
  SJF = 'sjf',
  PRIORITY = 'priority',
  RR = 'rr',
  EDF = 'edf'
}

三、资源分配

3.1 资源分配策略

在这里插入图片描述

图2：资源分配——静态分配、动态分配、预留分配、共享分配

策略描述优点缺点适用场景
静态分配固定资源分配简单可预测资源利用率低固定负载
动态分配按需动态调整资源利用率高实现复杂变化负载
预留分配预留部分资源保证关键任务资源浪费关键任务
共享分配资源共享使用最大化利用竞争问题通用场景

3.2 资源分配实现

// 资源分配器
class ResourceAllocator {
  private totalResources: Resources
  private allocatedResources: Map<string, Resources> = new Map()
  private resourcePool: ResourcePool

  constructor(totalResources: Resources) {
    this.totalResources = totalResources
    this.resourcePool = new ResourcePool(totalResources)
  }

  // 分配资源
  allocate(taskId: string, requirements: ResourceRequirements): boolean {
    // 检查资源是否充足
    if (!this.resourcePool.hasEnough(requirements)) {
      return false
    }

    // 分配资源
    const allocated = this.resourcePool.allocate(requirements)
    this.allocatedResources.set(taskId, allocated)

    return true
  }

  // 释放资源
  release(taskId: string): void {
    const resources = this.allocatedResources.get(taskId)
    if (resources) {
      this.resourcePool.release(resources)
      this.allocatedResources.delete(taskId)
    }
  }

  // 获取资源使用情况
  getResourceUsage(): ResourceUsage {
    const used = this.calculateUsedResources()
    
    return {
      total: this.totalResources,
      used,
      available: {
        cpu: this.totalResources.cpu - used.cpu,
        memory: this.totalResources.memory - used.memory,
        npu: this.totalResources.npu - used.npu
      },
      utilization: {
        cpu: used.cpu / this.totalResources.cpu,
        memory: used.memory / this.totalResources.memory,
        npu: used.npu / this.totalResources.npu
      }
    }
  }

  // 计算已使用资源
  private calculateUsedResources(): Resources {
    let used: Resources = { cpu: 0, memory: 0, npu: 0 }
    
    for (const [, resources] of this.allocatedResources) {
      used.cpu += resources.cpu
      used.memory += resources.memory
      used.npu += resources.npu
    }

    return used
  }
}

// 资源池
class ResourcePool {
  private available: Resources

  constructor(totalResources: Resources) {
    this.available = { ...totalResources }
  }

  // 检查资源是否充足
  hasEnough(requirements: ResourceRequirements): boolean {
    return this.available.cpu >= requirements.cpu &&
           this.available.memory >= requirements.memory &&
           this.available.npu >= requirements.npu
  }

  // 分配资源
  allocate(requirements: ResourceRequirements): Resources {
    this.available.cpu -= requirements.cpu
    this.available.memory -= requirements.memory
    this.available.npu -= requirements.npu

    return {
      cpu: requirements.cpu,
      memory: requirements.memory,
      npu: requirements.npu
    }
  }

  // 释放资源
  release(resources: Resources): void {
    this.available.cpu += resources.cpu
    this.available.memory += resources.memory
    this.available.npu += resources.npu
  }
}

// 资源
interface Resources {
  cpu: number
  memory: number
  npu: number
}

// 资源使用
interface ResourceUsage {
  total: Resources
  used: Resources
  available: Resources
  utilization: {
    cpu: number
    memory: number
    npu: number
  }
}

四、并发控制

4.1 并发控制机制

在这里插入图片描述

图3：并发控制——信号量、互斥锁、条件变量、原子操作

机制描述优点缺点适用场景
信号量计数器控制访问灵活易出错资源池
互斥锁独占访问简单可能死锁临界区
条件变量等待条件满足高效复杂同步
原子操作不可中断操作高效有限计数器

4.2 并发控制实现

// 并发控制器
class ConcurrencyController {
  private semaphore: Semaphore
  private mutex: Mutex
  private condition: ConditionVariable

  constructor(maxConcurrent: number) {
    this.semaphore = new Semaphore(maxConcurrent)
    this.mutex = new Mutex()
    this.condition = new ConditionVariable()
  }

  // 获取资源
  async acquire(): Promise<void> {
    await this.semaphore.acquire()
  }

  // 释放资源
  release(): void {
    this.semaphore.release()
  }

  // 执行临界区代码
  async executeCritical<T>(fn: () => T): Promise<T> {
    await this.mutex.lock()
    try {
      return fn()
    } finally {
      this.mutex.unlock()
    }
  }

  // 等待条件
  async waitForCondition(predicate: () => boolean): Promise<void> {
    while (!predicate()) {
      await this.condition.wait()
    }
  }

  // 通知条件
  notifyCondition(): void {
    this.condition.notify()
  }
}

// 信号量
class Semaphore {
  private count: number
  private waitQueue: (() => void)[] = []

  constructor(initialCount: number) {
    this.count = initialCount
  }

  // 获取
  async acquire(): Promise<void> {
    if (this.count > 0) {
      this.count--
      return
    }

    return new Promise(resolve => {
      this.waitQueue.push(resolve)
    })
  }

  // 释放
  release(): void {
    if (this.waitQueue.length > 0) {
      const resolve = this.waitQueue.shift()!
      resolve()
    } else {
      this.count++
    }
  }
}

// 互斥锁
class Mutex {
  private locked: boolean = false
  private waitQueue: (() => void)[] = []

  // 加锁
  async lock(): Promise<void> {
    if (!this.locked) {
      this.locked = true
      return
    }

    return new Promise(resolve => {
      this.waitQueue.push(resolve)
    })
  }

  // 解锁
  unlock(): void {
    if (this.waitQueue.length > 0) {
      const resolve = this.waitQueue.shift()!
      resolve()
    } else {
      this.locked = false
    }
  }
}

// 条件变量
class ConditionVariable {
  private waitQueue: (() => void)[] = []

  // 等待
  async wait(): Promise<void> {
    return new Promise(resolve => {
      this.waitQueue.push(resolve)
    })
  }

  // 通知
  notify(): void {
    if (this.waitQueue.length > 0) {
      const resolve = this.waitQueue.shift()!
      resolve()
    }
  }
}

五、性能监控

5.1 性能监控指标

在这里插入图片描述

图4：性能监控——CPU使用率、内存占用、任务延迟、吞吐量

指标描述目标值监控频率
CPU使用率CPU占用百分比< 80%实时
内存占用内存使用量< 80%实时
任务延迟任务完成时间< 100ms每次任务
吞吐量单位时间完成任务数> 100/s每秒
资源利用率资源使用效率> 70%每分钟

5.2 性能监控实现

// 性能监控器
class PerformanceMonitor {
  private metrics: PerformanceMetrics[] = []
  private isMonitoring: boolean = false

  // 开始监控
  startMonitoring(): void {
    this.isMonitoring = true
    this.monitoringLoop()
  }

  // 停止监控
  stopMonitoring(): void {
    this.isMonitoring = false
  }

  // 监控循环
  private async monitoringLoop(): Promise<void> {
    while (this.isMonitoring) {
      const metric = this.collectMetrics()
      this.metrics.push(metric)

      // 检查是否超出阈值
      this.checkThresholds(metric)

      await this.delay(1000)
    }
  }

  // 收集指标
  private collectMetrics(): PerformanceMetrics {
    return {
      timestamp: Date.now(),
      cpuUsage: this.getCPUUsage(),
      memoryUsage: this.getMemoryUsage(),
      taskLatency: this.getTaskLatency(),
      throughput: this.getThroughput()
    }
  }

  // 检查阈值
  private checkThresholds(metric: PerformanceMetrics): void {
    if (metric.cpuUsage > 80) {
      console.warn('CPU使用率超过阈值')
    }

    if (metric.memoryUsage > 80) {
      console.warn('内存使用率超过阈值')
    }

    if (metric.taskLatency > 100) {
      console.warn('任务延迟超过阈值')
    }
  }

  // 获取CPU使用率
  private getCPUUsage(): number {
    // 获取CPU使用率的实现
    return 0
  }

  // 获取内存使用率
  private getMemoryUsage(): number {
    // 获取内存使用率的实现
    return 0
  }

  // 获取任务延迟
  private getTaskLatency(): number {
    // 获取任务延迟的实现
    return 0
  }

  // 获取吞吐量
  private getThroughput(): number {
    // 获取吞吐量的实现
    return 0
  }

  // 延迟
  private delay(ms: number): Promise<void> {
    return new Promise(resolve => setTimeout(resolve, ms))
  }
}

// 性能指标
interface PerformanceMetrics {
  timestamp: number
  cpuUsage: number
  memoryUsage: number
  taskLatency: number
  throughput: number
}

六、实战案例

6.1 多任务AI应用

在这里插入图片描述

图5：多任务AI——图像识别、语音识别、NLP、推荐系统

// 多任务AI调度器
class MultiTaskAIScheduler {
  private taskScheduler: TaskScheduler
  private resourceAllocator: ResourceAllocator
  private concurrencyController: ConcurrencyController

  constructor() {
    this.taskScheduler = new TaskScheduler('priority')
    this.resourceAllocator = new ResourceAllocator({
      cpu: 8,
      memory: 16,
      npu: 2
    })
    this.concurrencyController = new ConcurrencyController(4)
  }

  // 提交任务
  async submitTask(task: AITask): Promise<void> {
    // 添加任务到调度队列
    this.taskScheduler.addTask(task)

    // 尝试分配资源
    const allocated = this.resourceAllocator.allocate(task.id, task.resourceRequirements)

    if (!allocated) {
      console.warn(`无法为任务 ${task.name} 分配资源`)
      return
    }

    // 获取并发许可
    await this.concurrencyController.acquire()

    try {
      // 执行任务
      await this.executeTask(task)
    } finally {
      // 释放资源
      this.resourceAllocator.release(task.id)
      this.concurrencyController.release()
    }
  }

  // 执行任务
  private async executeTask(task: AITask): Promise<void> {
    console.log(`执行任务: ${task.name}`)
    
    // 模拟任务执行
    await this.delay(task.estimatedTime)
    
    console.log(`任务完成: ${task.name}`)
  }

  // 延迟
  private delay(ms: number): Promise<void> {
    return new Promise(resolve => setTimeout(resolve, ms))
  }
}

七、常见问题与解决

问题现象原因解决方案
任务饥饿低优先级任务无法执行高优先级任务过多优先级老化、时间片分配
资源死锁任务互相等待资源资源分配不当资源排序、超时释放
性能抖动性能不稳定调度策略不当动态调整、负载均衡
内存泄漏内存持续增长资源未释放自动释放、内存监控
实时性不足实时任务延迟高调度延迟优先级抢占、快速调度
功耗过高设备发热严重任务调度不合理功耗感知调度、动态频率

八、结语：多任务调度是端侧AI的"大脑"

多任务调度是端侧AI的"大脑"，协调各种任务的执行：

  • 任务调度：合理安排任务执行顺序
  • 资源分配：高效利用有限的资源
  • 并发控制：保证任务正确执行
  • 性能监控：实时监控系统状态

作为一名讲师，我在课上常说：**"好的调度器，让端侧AI像交响乐团一样和谐。"**

转载自：https://blog.csdn.net/u014727709/article/details/165126824
欢迎 👍点赞✍评论⭐收藏，欢迎指正

Logo
HarmonyOS开发者社区

讨论HarmonyOS开发技术，专注于API与组件、DevEco Studio、测试、元服务和应用上架分发等。

更多推荐

cover

ToDesk鸿蒙v4.8.1.0首发实测：大屏远控+互动板块+手势指南+实时响应，四项升级值不值？

avatar HarmonyOS开发者社区
cover

HarmonyOS 7 ContainerReader 实战：组件为什么也需要自己的响应式断点【鸿蒙心迹】

avatar HarmonyOS开发者社区
cover

HarmonyOS 7 长列表组件复用：从 @ReusableV2 到全局复用池【鸿蒙心迹】

avatar HarmonyOS开发者社区