Defrag/job.py at main · IntelligentDDS/Defrag · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106

class Job(dict):
    def __init__(self, series):
        super(Job, self).__init__()
        self.update(series.to_dict())
        self.update({"nodes": [], "priority": -1, "random": 0})
        self.update({"profiled": 0, "profqueue": 0, "toskip": 0})
        self.update({"exclusive": 1, "rate": 1, "sharescore": None, "Tcolocate": 0, "Tdelocate": 0})
        self.update({"unschedulabletime": 0, "retry": 0, "backofftime": 0})
        self.update({"overheadtime": 0})

    def set_ckpt_time(self, time):
        self.last_ckpt_time = time

    def get_ckpt_time(self):
        return self.last_ckpt_time

    def __hash__(self):
        return hash(self.get('job_id'))

    def __eq__(self, other):
        if isinstance(other, Job):
            return self.get('job_id') == other.get('job_id')
        return False


class TargetJob:
    def __init__(self, cpu_num, gpu_num, node_num):
        self.cpu_num = cpu_num
        self.gpu_num = gpu_num
        self.node_num = node_num

    def __eq__(self, other):
        if isinstance(other, TargetJob):
            return (self.cpu_num == other.cpu_num and
                    self.gpu_num == other.gpu_num and
                    self.node_num == other.node_num)
        return False

    def __hash__(self):
        return hash((self.cpu_num, self.gpu_num, self.node_num))


class Trace:
    def __init__(self, experiment_name, args):
        self.experiment_name = experiment_name
        self.job_list = []
        self.job_list2 = [] # for FGD
        self.typical_jobs = []
        self.args = args

    def append_job(self, job):
        self.job_list.append(job)

    def append_job2(self, job):
        self.job_list2.append(job)

    def job_num(self):
        return len(self.job_list)

    def profiler_remain_job_num(self):
        num = 0
        for job in self.job_list:
            if job["toskip"] == 0:
                num += 1
        return num

    def sort_jobs(self, key):
        self.job_list.sort(key=lambda x: x.__getitem__(key))

    def reset_trace(self, CLUSTER):
        if "PAI" in self.experiment_name:
            for job in self.job_list:
                if job["gpu_num"] / 100 > CLUSTER.num_gpus_per_node or job["cpu_num"] / 100 > CLUSTER.num_cpus_per_node \
                        or job["node_num"] > CLUSTER.node_num:
                    job["toskip"] = 1
                if job["cpu_num"] == 0:
                    job["toskip"] = 1
                job["node_num"] = 1

        if self.args.scheduler == "FGD":
            typical_dict = {}
            total_job = 0
            for job in self.job_list2:
                if job["toskip"] == 0:
                    total_job += 1
                    tmp_job = TargetJob(job["cpu_num"], job["gpu_num"], job["node_num"])
                    if tmp_job not in typical_dict:
                        typical_dict[tmp_job] = 1
                    else:
                        typical_dict[tmp_job] += 1

            new_total_job = total_job
            to_remove = []
            for k in typical_dict:
                percentage = typical_dict[k] / total_job
                if percentage < self.args.fgd_scaling:
                    to_remove.append(k)
            for k in to_remove:
                del typical_dict[k]

            for k in typical_dict:
                percentage = typical_dict[k] / new_total_job
                self.typical_jobs.append({"TargetJob": k, "Percentage": percentage})

            self.typical_jobs.sort(key=lambda x: x["Percentage"], reverse=True)