LoadBalancer/solve_additional.cpp at main · DrAsu33/LoadBalancer · 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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
#include "servercluster.h"

// 迁移计划的数据结构
struct MigrationPlan
{
    size_t task_id;
    size_t from_node;
    size_t to_node;
    int demand;
    bool is_completed;
    std::vector<size_t> path_nodes;
    MigrationPlan(size_t id, size_t f, size_t t, int d, const std::vector<size_t>& p)
        : task_id(id), from_node(f), to_node(t), demand(d), is_completed(false), path_nodes(p) {}
};

//每一步任务移动信息的数据结构
struct StepLog
{
    int timestep;
    size_t task_id, from, to;
    StepLog(int t, size_t tid, size_t f, size_t t1) : timestep(t), task_id(tid), from(f), to(t1) {}
    void print_log() const {std::cout << timestep << " " << task_id << " " << from << " " << to << std::endl;}
};

// 启动优化函数， 该函数是附加要求的函数
void ServerCluster::solve_additional()
{

    // 先使用floyd算法检测每两个服务器是否连通, 并储存最短路径, 并建立log向量
    run_floyd_parent();
    std::vector<StepLog> logs;
    int timestep = 0;
    std::vector<MigrationPlan> pending_queue;

    // 贪心策略把所有需要的迁移计划加入队列
    for(auto& server : servers)
    {
        // 该服务器没有超载，不需要动
        if(server.gpu_used <= server.capacity || server.assigned_tasks.empty())
            continue;

        int overflow = server.gpu_used - server.capacity;
        // 按照任务的要求负载从大到小排序 从大到小传输走任务
        std::sort(server.assigned_tasks.begin(), server.assigned_tasks.end(),
            [this](size_t task_idx_a, size_t task_idx_b)
            {
                return this->tasks[task_idx_a].demand > this->tasks[task_idx_b].demand;
            }
        );

        auto it = server.assigned_tasks.begin();
        while(it != server.assigned_tasks.end() && overflow > 0)
        {
            size_t task_idx = *it;
            Task& task = tasks[task_idx];
            bool moved = false;
            // 用自己作为初值，如果操作完minserver还是自己说明没找到
            size_t best_dest_index = server.index; long long min_cost = INF;
            // 尝试将该任务迁移到其他服务器
            for(size_t dest_idx = 0; dest_idx < servers.size(); dest_idx++)
            {
                // 不能传输给自己和不连通的服务器
                if(server.index == dest_idx || adjacency_matrix[server.index][dest_idx] == INF)
                    continue;

                Server& dest_server = servers[dest_idx];
                // 可以容纳该任务,如果路径更短则更新
                if(dest_server.capacity >= task.demand + dest_server.gpu_used)
                {
                    if(adjacency_matrix[server.index][dest_idx] < min_cost)
                    {
                        min_cost = adjacency_matrix[server.index][dest_idx];
                        best_dest_index = dest_idx;
                    }
                }
            }
            // 把该传输计划存入队列
            if(best_dest_index != server.index)
            {
                pending_queue.push_back(MigrationPlan(task_idx, server.index, best_dest_index, task.demand, get_path(server.index, best_dest_index)));
                server.gpu_used -= task.demand;
                overflow -= task.demand;
                servers[best_dest_index].gpu_used += task.demand;

                it = server.assigned_tasks.erase(it);
                servers[best_dest_index].assigned_tasks.push_back(task_idx);
                task.current_node = best_dest_index;
                task.migration_cost += min_cost * task.demand;
                moved = true;
            }
            if(!moved)
                it++;
        }
    }

    size_t count = 0; // 记录已经迁移的任务数量
    // 一直循环到所有任务迁移结束
    while(count < pending_queue.size())
    {
        timestep++; // 时间递增， 从1开始
        auto current_bandwidth = bandwidth_matrix;  // 每次循环都进行一次深拷贝， 考虑到n不会超过50所以可以接受
        bool deadlock_check = false; // 检测是否产生死锁，导致死循环
        for(auto& plan : pending_queue)
        {
            // 如果任务已经迁移结束则跳过
            if(plan.is_completed)
                continue;

            bool enable = true;  // 是否有足够的带宽迁移
            const std::vector<size_t>& path = plan.path_nodes;
            // 防止因意外导致数组越界访问
            if(path.size() < 2)
                continue;
            for(size_t i = 0; i < path.size() - 1; i++)
            {
                if(current_bandwidth[path[i]][path[i+1]] < 1)
                    enable = false;
            }
            // 如果带宽不足就跳过
            if(!enable)
                continue;

            count++;
            // 依次消耗带宽
            for(size_t i = 0; i < path.size() - 1; i++)
                current_bandwidth[path[i]][path[i+1]] -= 1;
            plan.is_completed = true;
            deadlock_check = true;
            logs.push_back(StepLog(timestep, plan.task_id + 1, plan.from_node + 1, plan.to_node + 1));
        }
        if (!deadlock_check && count < pending_queue.size())
        {
            std::cerr << "Error: Deadlock detected at timestep " << timestep << "! Bandwidth insufficient." << std::endl;
            break;
        }
    }
    long long total_cost = 0;
    // 输出迁移所需要的总时间
    std::cout << timestep << std::endl;
    // 输出所有记录下来的移动信息
    for(const auto& log : logs)
        log.print_log();

    // 输出每个任务的迁移结果
    for(const auto& task : tasks)
    {
        std::cout << task.index + 1 << " " << task.start_node + 1 << " " <<
        task.current_node  + 1 << " " << task.migration_cost << std::endl;
        total_cost += task.migration_cost;
    }
    // 输出每台服务器最终负载
    for(const auto& server : servers)
        std::cout << server.index + 1 << " " << server.gpu_used << std::endl;
    // 输出总共的迁移成本
    std::cout << total_cost << std::endl;
}