华为非AI方向笔试真题-昇腾NPU协同调度系统(详细思路+多语言题解)

知识不进脑子的人

40人浏览 · 2026-06-06 12:05:20

知识不进脑子的人 · 2026-06-06 12:05:20 发布

昇腾NPU协同调度系统

华为笔试真题 6月3号非AI方向 300分题型

题目内容

设计一个调度系统管理昇腾 $N P U$ 上的训练任务，需满足以下条件：

任务属性（每个任务包含 $3$ 个维度）

$n p u s$ ：所需 NPU 卡编号列表
$d u r a t i o n$ ：执行需要的时间片数
$d e p e n d o n$ ：必须在该任务前完成的任务 $I D$ （只会依赖 $0$ 或者 $1$ 个上游任务）
调度规则
每张 $N P U$ 卡同一时间只能执行一个任务
任务必须在前置任务完成后才能开始
时间片为正整数，多个满足条件的任务可并行启动，不可抢占已分配任务
可能存在多种调度方案，需要选出全部任务结束耗时最短的一种
用例保证存在可调度方案（不存在任务间循环依赖），输出所有任务全部完成需要的最短时间。

输入描述

第 $1$ 行， $N P U$ 卡总数（ $1 <= N P U 卡总数 <= 10$ ），总任务数量（ $1 <= 总任务数量 <= 8$ ），如：4 2
第 $2$ 行，【第 $2 - 4$ 行第 $0$ 个任务】，任务 $0$ 依赖的 $n p u$ 列表（空格分割），如： $0$ $1$
第 $3$ 行，【第 $2 - 4$ 行第 $0$ 个任务】，任务 $0$ 持续时间，如： $5$
第 $4$ 行，【第 $2 - 4$ 行第 $0$ 个任务】，任务 $0$ 依赖的上游任务 $I D$ （无依赖时固定为 $- 1$ ），如： $1$
第 $5$ 行，【第 $5 - 7$ 行第 $1$ 个任务】，任务 $1$ 依赖的 $n p u$ 列表（空格分割），如： $1$ $2$
第 $6$ 行，【第 $5 - 7$ 行第 $1$ 个任务】，任务 $1$ 持续时间，如： $4$
第 $7$ 行，【第 $5 - 7$ 行第 $1$ 个任务】，任务 $1$ 依赖的上游任务 $I D$ （无依赖时固定为 $- 1$ ），如： $0$
……，如果存在更多任务， $3$ 行一组进行输入，任务 $i$ 依赖的 $n p u$ 列表（空格分割），如： $1$ $2$
……，如果存在更多任务， $3$ 行一组进行输入，任务 $i$ 持续时间，如： $4$
……，如果存在更多任务， $3$ 行一组进行输入，任务 $i$ 依赖的上游任务 $I D$ （无依赖时固定为 $- 1$ ），如： $0$
注意：题目保证输入合法，无需校验输入

输出描述

格式： $in t$ 类型数值
解释：所有任务全部完成需要的最短时间，用例保证存在可调度方案（不存在任务间循环依赖）

样例1

输入

输出

说明

$2$ # $2$ 个NPU卡， $3$ 个任务
$0$ # 任务 $0$ 需要NPU卡 $0$
$3$ # 任务 $0$ 持续 $3$
$- 1$ # 任务 $0$ 无依赖
$0$ # 任务 $1$ 需要NPU卡 $0$
$2$ # 任务 $1$ 持续 $2$
$- 1$ # 任务 $1$ 无依赖
$1$ # 任务 $2$ 需要NPU卡 $1$
$4$ # 任务 $2$ 持续 $4$
$1$ # 任务 $2$ 依赖任务 $1$
方案1：优先执行耗时长的任务（任务 $0$ 先执行）
时间 $0$ ~ $3$ NPU $0$ : 任务 $0$ （持续 $3$ ，时间 $0$ - $3$ ） NPU $1$ : 空闲（任务 $2$ 依赖任务 $1$ ，还未完成）
时间 $3$ ~ $5$ NPU $0$ : 任务 $1$ （持续 $2$ ，时间 $3$ - $5$ ） NPU $1$ : 空闲（任务 $2$ 需等待任务 $1$ ，任务 $1$ 在时间 $5$ 完成，但任务 $2$ 还未开始）
时间 $5$ ~ $9$ NPU $0$ : 空闲 NPU $1$ : 任务 $2$ （持续 $4$ ，时间 $5$ - $9$ ）
任务完成时间： 任务 $0$ : 时间 $3$ 完成任务 $1$ : 时间 $5$ 完成任务 $2$ : 时间 $9$ 完成
总时间： $9$
方案2：优先执行耗时短的任务（任务 $1$ 先执行）
时间 $0$ ~ $2$ NPU $0$ : 任务 $1$ （持续 $2$ ，时间 $0$ - $2$ ） NPU $1$ : 空闲（任务 $2$ 依赖任务 $1$ ，还未完成）
时间 $2$ ~ $6$ NPU $0$ : 任务 $0$ （持续 $3$ ，时间 $2$ - $5$ ） NPU $1$ : 任务 $2$ （持续 $4$ ，时间 $2$ - $6$ ，任务 $1$ 在时间 $2$ 已完成，满足依赖）
时间 $5$ ~ $6$ NPU $0$ : 空闲（任务 $0$ 已完成） NPU $1$ : 任务 $2$ 继续执行（时间 $5$ - $6$ ）
任务完成时间：
任务 $1$ : 时间 $2$ 完成任务 $0$ : 时间 $5$ 完成任务 $2$ : 时间 $6$ 完成
总时间： $6$

样例2

输入

输出

说明

$3$ $2$ # $3$ 个NPU卡， $2$ 个任务
$10\ 1$ # 任务 $0$ 需要NPU卡 $0$ 、卡 $1$
$4$ # 任务 $0$ 持续 $4$
$- 1$ # 任务 $0$ 无依赖
$21\ 2$ # 任务 $1$ 需要NPU卡 $1$ 、卡 $2$
$5$ # 任务 $1$ 持续 $5$
$0$ # 任务 $1$ 依赖任务 $0$
调度顺序：任务 $1$ 依赖任务 $0$ ，只能先执行任务 $0$ 再执行任务 $1$ ，总耗时 $4 + 5 = 9$ ，输出 $9$
提示 $N P U$ 编号从 $0$ 开始，任务编号从 $O$ 开始

题解

思路

递归回溯实现，由于本题任务数较小1 <= M <= 8完成可以枚举所有执行情况来确定最小时间。具体逻辑如下：

接受输入，记录每个任务的依赖NPU,任务执行时长和依赖任务
定义finish记录每个任务完成时间，npuAvalid记录每个NPU空闲时间，complete二进制记录每个任务完成标志。然后进行递归回溯
递归从前往后选取一个任务执行，选取任务需要满足以下条件
- 任务还未完成
- 任务依赖的任务已经完成
计算任务的开始时间startTime，为依赖的npu空闲时间和依赖任务的最大值，任务结束时间就为startTime + duration，更新对应finish以及依赖Npu的空闲时间，往下进行递归
递归结束条件为complete == (1 << m) -1

C++

#include<bits/stdc++.h>
using namespace std;

struct Task {
    vector<int> npus;
    int duration;
    int pre;
};


int ans = INT_MAX;
// NPU卡 // 任务数量
int m, n;
vector<Task> task;
vector<int> finish;

void dfs(int complete, vector<int>& npuAvail) {
    // 所有任务都已调度，更新最优解
    if (complete == (1 << n) - 1) {
        int finishTime = *max_element(npuAvail.begin(), npuAvail.end());
        ans = min(ans, finishTime);
        return;
    }
    // 剪枝：当前时间已超过已知最优解
    int curMax = *max_element(npuAvail.begin(), npuAvail.end());
    if (curMax >= ans) return;
    // 尝试调度一个任务
    for (int i = 0; i < n; i++) {
        // 跳过已完成的任务
        if (complete & (1 << i)) continue;
        // 检查依赖是否满足
        if (task[i].pre != -1 && !(complete & (1 << task[i].pre))) {
            continue;
        }
        // 计算任务最早开始时间
        int startTime = 0;
        for (int npu : task[i].npus) {
            startTime = max(startTime, npuAvail[npu]);
        }
        // 晚于依赖任务完成时间
        if (task[i].pre != -1) {
            startTime = max(startTime, finish[task[i].pre]);
        }
        int endTime = startTime + task[i].duration;
        // 记录NPU的原始状态，用于回溯
        vector<int> oldAvail(m);
        for (int npu : task[i].npus) {
            oldAvail[npu] = npuAvail[npu];
            npuAvail[npu] = endTime;  // 分配NPU
        }
        finish[i] = endTime;
        dfs(complete | (1 << i), npuAvail);
        finish[i] = -1;
        // 回溯：恢复NPU状态
        for (int npu : task[i].npus) {
            npuAvail[npu] = oldAvail[npu];
        }
    }
}


int main() {
    ios::sync_with_stdio(false);
    cin.tie(nullptr);
    cin >> m >> n; 
    task.resize(n);
    for (int i = 0; i < n; i++) {
        string line;
        cin.ignore();
        getline(cin, line);
        stringstream ss(line);
        int x;
        while (ss >> x) {
            task[i].npus.push_back(x);
        }
        cin >> task[i].duration;
        cin >> task[i].pre;
    }
    
    finish.resize(n, -1);
    vector<int> npuAvail(m, 0);
    dfs(0, npuAvail);


    cout << ans << "\n";
}

java

import java.io.*;
import java.util.*;

public class Main {

    static class Task {
        List<Integer> npus = new ArrayList<>();
        int duration;
        int pre;
    }

    static int ans = Integer.MAX_VALUE;

    // NPU卡数量 // 任务数量
    static int m, n;

    static List<Task> task = new ArrayList<>();
    static int[] finish;

    static void dfs(int complete, int[] npuAvail) {
        // 所有任务都已调度，更新最优解
        if (complete == (1 << n) - 1) {
            int finishTime = 0;
            for (int x : npuAvail) {
                finishTime = Math.max(finishTime, x);
            }
            ans = Math.min(ans, finishTime);
            return;
        }

        // 剪枝：当前时间已超过已知最优解
        int curMax = 0;
        for (int x : npuAvail) {
            curMax = Math.max(curMax, x);
        }
        if (curMax >= ans) return;

        // 尝试调度一个任务
        for (int i = 0; i < n; i++) {

            // 跳过已完成的任务
            if ((complete & (1 << i)) != 0) continue;

            // 检查依赖是否满足
            if (task.get(i).pre != -1 &&
                    (complete & (1 << task.get(i).pre)) == 0) {
                continue;
            }

            // 计算任务最早开始时间
            int startTime = 0;

            for (int npu : task.get(i).npus) {
                startTime = Math.max(startTime, npuAvail[npu]);
            }

            // 晚于依赖任务完成时间
            if (task.get(i).pre != -1) {
                startTime = Math.max(startTime, finish[task.get(i).pre]);
            }

            int endTime = startTime + task.get(i).duration;

            // 记录NPU原状态
            int[] oldAvail = new int[m];

            for (int npu : task.get(i).npus) {
                oldAvail[npu] = npuAvail[npu];
                npuAvail[npu] = endTime;
            }

            finish[i] = endTime;

            dfs(complete | (1 << i), npuAvail);

            finish[i] = -1;

            // 回溯
            for (int npu : task.get(i).npus) {
                npuAvail[npu] = oldAvail[npu];
            }
        }
    }

    public static void main(String[] args) throws Exception {

        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));

        String[] first = br.readLine().trim().split("\\s+");

        m = Integer.parseInt(first[0]);
        n = Integer.parseInt(first[1]);

        for (int i = 0; i < n; i++) {

            Task t = new Task();

            String[] npuLine = br.readLine().trim().split("\\s+");

            for (String s : npuLine) {
                t.npus.add(Integer.parseInt(s));
            }

            t.duration = Integer.parseInt(br.readLine().trim());
            t.pre = Integer.parseInt(br.readLine().trim());

            task.add(t);
        }

        finish = new int[n];
        Arrays.fill(finish, -1);

        int[] npuAvail = new int[m];

        dfs(0, npuAvail);

        System.out.println(ans);
    }
}

python

import sys

ans = float('inf')

# NPU卡数量 // 任务数量
m, n = map(int, input().split())

task = []

for _ in range(n):
    npus = list(map(int, input().split()))
    duration = int(input())
    pre = int(input())

    task.append({
        "npus": npus,
        "duration": duration,
        "pre": pre
    })

finish = [-1] * n


def dfs(complete, npu_avail):
    global ans

    # 所有任务都已调度，更新最优解
    if complete == (1 << n) - 1:
        finish_time = max(npu_avail)
        ans = min(ans, finish_time)
        return

    # 剪枝：当前时间已超过已知最优解
    cur_max = max(npu_avail)
    if cur_max >= ans:
        return

    # 尝试调度一个任务
    for i in range(n):

        # 跳过已完成的任务
        if complete & (1 << i):
            continue

        # 检查依赖是否满足
        if task[i]["pre"] != -1 and not (complete & (1 << task[i]["pre"])):
            continue

        # 计算任务最早开始时间
        start_time = 0

        for npu in task[i]["npus"]:
            start_time = max(start_time, npu_avail[npu])

        # 晚于依赖任务完成时间
        if task[i]["pre"] != -1:
            start_time = max(start_time, finish[task[i]["pre"]])

        end_time = start_time + task[i]["duration"]

        old_avail = {}

        for npu in task[i]["npus"]:
            old_avail[npu] = npu_avail[npu]
            npu_avail[npu] = end_time

        finish[i] = end_time

        dfs(complete | (1 << i), npu_avail)

        finish[i] = -1

        # 回溯
        for npu in task[i]["npus"]:
            npu_avail[npu] = old_avail[npu]


npu_avail = [0] * m

dfs(0, npu_avail)

print(ans)

javascript

const readline = require('readline');

const rl = readline.createInterface({
    input: process.stdin,
    output: process.stdout
});

const lines = [];

rl.on('line', line => {
    lines.push(line.trim());
});

rl.on('close', () => {

    let idx = 0;

    const [m, n] = lines[idx++].split(' ').map(Number);

    const task = [];

    for (let i = 0; i < n; i++) {

        const npus = lines[idx++].split(' ').map(Number);

        const duration = Number(lines[idx++]);

        const pre = Number(lines[idx++]);

        task.push({
            npus,
            duration,
            pre
        });
    }

    let ans = Number.MAX_SAFE_INTEGER;

    const finish = new Array(n).fill(-1);

    function dfs(complete, npuAvail) {

        // 所有任务都已调度，更新最优解
        if (complete === (1 << n) - 1) {
            const finishTime = Math.max(...npuAvail);
            ans = Math.min(ans, finishTime);
            return;
        }

        // 剪枝：当前时间已超过已知最优解
        const curMax = Math.max(...npuAvail);

        if (curMax >= ans) {
            return;
        }

        // 尝试调度一个任务
        for (let i = 0; i < n; i++) {

            // 跳过已完成的任务
            if (complete & (1 << i)) continue;

            // 检查依赖是否满足
            if (
                task[i].pre !== -1 &&
                !(complete & (1 << task[i].pre))
            ) {
                continue;
            }

            // 计算任务最早开始时间
            let startTime = 0;

            for (const npu of task[i].npus) {
                startTime = Math.max(startTime, npuAvail[npu]);
            }

            // 晚于依赖任务完成时间
            if (task[i].pre !== -1) {
                startTime = Math.max(
                    startTime,
                    finish[task[i].pre]
                );
            }

            const endTime = startTime + task[i].duration;

            const oldAvail = {};

            for (const npu of task[i].npus) {
                oldAvail[npu] = npuAvail[npu];
                npuAvail[npu] = endTime;
            }

            finish[i] = endTime;

            dfs(complete | (1 << i), npuAvail);

            finish[i] = -1;

            // 回溯
            for (const npu of task[i].npus) {
                npuAvail[npu] = oldAvail[npu];
            }
        }
    }

    const npuAvail = new Array(m).fill(0);

    dfs(0, npuAvail);

    console.log(ans);
});

Go

package main

import (
	"bufio"
	"fmt"
	"os"
	"strconv"
	"strings"
)

type Task struct {
	npus     []int
	duration int
	pre      int
}

var ans = int(^uint(0) >> 1)

// NPU卡数量 // 任务数量
var m, n int

var task []Task
var finish []int

func max(a, b int) int {
	if a > b {
		return a
	}
	return b
}

func dfs(complete int, npuAvail []int) {

	// 所有任务都已调度，更新最优解
	if complete == (1<<n)-1 {

		finishTime := 0

		for _, v := range npuAvail {
			if v > finishTime {
				finishTime = v
			}
		}

		if finishTime < ans {
			ans = finishTime
		}

		return
	}

	// 剪枝：当前时间已超过已知最优解
	curMax := 0

	for _, v := range npuAvail {
		if v > curMax {
			curMax = v
		}
	}

	if curMax >= ans {
		return
	}

	// 尝试调度一个任务
	for i := 0; i < n; i++ {

		// 跳过已完成的任务
		if complete&(1<<i) != 0 {
			continue
		}

		// 检查依赖是否满足
		if task[i].pre != -1 &&
			(complete&(1<<task[i].pre)) == 0 {
			continue
		}

		// 计算任务最早开始时间
		startTime := 0

		for _, npu := range task[i].npus {
			startTime = max(startTime, npuAvail[npu])
		}

		// 晚于依赖任务完成时间
		if task[i].pre != -1 {
			startTime = max(startTime, finish[task[i].pre])
		}

		endTime := startTime + task[i].duration

		oldAvail := make(map[int]int)

		for _, npu := range task[i].npus {
			oldAvail[npu] = npuAvail[npu]
			npuAvail[npu] = endTime
		}

		finish[i] = endTime

		dfs(complete|(1<<i), npuAvail)

		finish[i] = -1

		// 回溯
		for _, npu := range task[i].npus {
			npuAvail[npu] = oldAvail[npu]
		}
	}
}

func main() {

	reader := bufio.NewReader(os.Stdin)

	line, _ := reader.ReadString('\n')
	line = strings.TrimSpace(line)

	first := strings.Fields(line)

	m, _ = strconv.Atoi(first[0])
	n, _ = strconv.Atoi(first[1])

	task = make([]Task, n)

	for i := 0; i < n; i++ {

		line, _ = reader.ReadString('\n')
		line = strings.TrimSpace(line)

		fields := strings.Fields(line)

		var npus []int

		for _, s := range fields {
			x, _ := strconv.Atoi(s)
			npus = append(npus, x)
		}

		line, _ = reader.ReadString('\n')
		duration, _ := strconv.Atoi(strings.TrimSpace(line))

		line, _ = reader.ReadString('\n')
		pre, _ := strconv.Atoi(strings.TrimSpace(line))

		task[i] = Task{
			npus:     npus,
			duration: duration,
			pre:      pre,
		}
	}

	finish = make([]int, n)

	for i := range finish {
		finish[i] = -1
	}

	npuAvail := make([]int, m)

	dfs(0, npuAvail)

	fmt.Println(ans)
}

AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

生成式 AI 赋能下新型网络钓鱼攻击机理与全链路防御研究

AtomGit开源社区

Gemini 怎么导出 pdf？实用小工具 AI 导出鸭上线，简单操作就能高效导出规范 PDF 文件

AtomGit开源社区

debugpy：微软开源的 Python 调试器

微软开源的Python调试器debugpy提供了灵活可靠的调试方案，支持命令行和代码嵌入两种使用模式。它实现了Debug Adapter Protocol，可在本地或远程进行调试，特别适合服务器问题排查和容器化环境。工具支持断点设置、子进程调试和进程附加功能，文档齐全且采用MIT协议开源。debugpy既可作为独立调试器使用，也能与支持DAP的编辑器集成，为Python开发者提供了不依赖特定IDE