一，題目
Instruction:
Please
complete the following task. Use whatever tools you find appropriate
(using the Internet, for example, is assumed). Linux, C++, Python,
CMake, Qt, and OpenCV are preferred for development but you may use
other tools if you are not familiar with these. Please include the
following support material
A
discussion of your approach and all the assumptions and tradeoffs
that you made
Any
build configuration files and instructions for compiling
A
list of any packages/utilities required to build and execute your
solution
Any
test code, or test data sets you generated
References to other
materials as appropriate
Please
do not spend more than 8‐10 hours on this task and feel free to ask
questions. While we would love to see a completed product, we expect
that your final solution may not be completely “finished” in the
available time. We are more interested in seeing how you approach a
problem, how you manage the deadline, how you communicate about your
work, and the quality of your work product.
Task: Build a planner
Engineers
are expected to be able to reason about large systems and the complex
interactions they may have. We need to be able to write software that
will coordinate and manage a variety of different tasks.
Please
write software that will tackle the following problem. Use whatever
tools you find appropriate, Please document you work, discuss your
approach, and enumerate all the assumptions and trade-offs that you
made. Please do not spend more than 8-10 hours on this, We are most
interested in seeing how you approach a problem, how you manage the
deadline, how you communicate your thoughts around the problem, and
what your view of “quality” is. We wish you to submit your
software, instructions on how to run it, an example data set that we
can run, and documentation on any design decisions or assumptions you
made.
Imagine
you are preparing a system to be able to coordinate the execution of
a large number of tasks. Many of these tasks are dependent on each
other and all have different requirements. As resources, you have a
heterogeneous cluster of computers. You wish to exploit parallelism
and complete the entire set of tasks as quickly as possible.
We
will define a list of tasks, and their dependencies, in a YAML file
that looks like so:
task1:
cores_required: 2
execution_time: 100
task2:
cores_required: 1
execution_time: 200
parent_tasks: task1
task3:
cores_required: 4
execution_time: 50
parent_tasks: “task1,
task2”
cores_required
means that the all the cores must be available on the same computing
resource, execution_time is the number of global “ticks” that
pass before the task will be complete and can release all its
computing resources, and the parent_tasks must be completely finished
executing before a particular task can start.
We
will then also specify the computing resources and their number of
available cores as so:
compute1:
2
compute2:
2
compute3:
6
Please
implement a program that reads those two files, and then emits an
ordered scheduling plan that manages all the dependencies (execution
time, number of cores, parent tasks). The goal is to execute the full
lists of tasks as quickly as possible.
The
plan may look something like:
task1:
compute1
task2:
compute1
task3:
compute3
二，解答
1.這個任務的時間規劃如下：
（1）30分鐘分析這個任務，包括調研一些可能使用到一些技術方案，例如這個裡面涉及到YAML檔案的解析，通過網絡需要找到一個可用的庫并簡單demo驗證；
（2）使用6小時30分鐘進行代碼編寫和測試；
（3）最後一個小時做一些文檔整理和軟體釋出準備；
（4）如果其中有些計劃延遲，可以在多出一點時間來修補，這個必須要控制在兩個小時以内。
其中第二步的代碼編寫和測試有最大的不可控，特别是在完成任務排程算法的時候。
2.下面在說說自己完成這個任務的一些思路
（1）解析任務：第一遍完整的看完這個題目，想到的就是一個任務排程功能，由于平時自己接觸過一些任務排程，例如作業系統裡面的任務排程，還有很多開源的任務排程架構，包括hadoop裡面的最新yarn也涉及到業務排程。是以第一個想法就是可不可以找一個比較好的開源任務排程架構來完成這個功能，但是仔細一想，這個題目也自己特殊的要求，例如從配置檔案讀取任務清單和計算資源池，而且任務還有依賴。那麼我的第二個想法是能不能找一個簡單的架構進行改進來完成，其實大部分架構都是比較複雜，如果要改造那麼需要對架構有深入的掌握，這個可能也不是在8-10個小時内能夠完成的。是以經過簡單想法驗證以後，我還是決定了從0開始自己實作，畢竟是一個功能比較單一的任務。不過下這個決定的時候自己也不清楚在有限的時間内能夠完成到什麼樣的程度，不過我想基本的功能應該實作是沒有問題的，因為自己以前了解過大部分的任務排程算法，不過肯定達不到最好的産品和最好的算法（性能）的要求。
（2）分解任務：我首先簡單把這個任務分解成兩個子任務，一個是解析配置檔案，二是實作具體的排程算法。第一個比較簡單，也很容易驗證測試；第二個剛開始也認為不是很難，不過後面在考慮這個情況和測試的時候會遇到很多問題。
（3）實作任務：因為分解成了兩個任務，是以先實作第一個任務。第一個任務涉及到兩種不同配置檔案解析。第一個是yaml格式的，因為yaml格式以前沒有接觸過，但是我想既然是一種通用的格式，應該很容易找到開源的實作，經過簡單的查找和驗證，snakeyaml這個開源庫很容易實作了解析yaml格式的需求。至于第二個檔案格式，一看就有點像是鍵值對的實作，簡單一查java自帶的就有。
第二個任務，也就是實作任務排程算法。從配置檔案拿到了任務清單和計算資源池，那麼就可以根據這些進行任務排程了。這個算法給人第一映像很簡單，不就是給任務配置設定合适的資源嘛。但是真正開始構思算法的時候才發現有很多情況需要考慮。先說一下異常情況，就是不能完成排程的情況。題目有一個限制就是一個任務必須一次性在一個計算資源上進行配置設定，那麼就存在這樣一個情況，如果有一個任務的計算資源很大，沒有一個計算資源能夠提供，那麼肯定就完不成；第二種情況就是如果任務存在這循環依賴，肯定也是完不成的，檢查循環依賴就是附加出來的一個算法，這個是前期考慮不足的，是以在計劃時間外，那麼在有限的時間裡面，肯定要重新調整一下。在實作計算資源是否滿足需要和驗證任務循環依賴，我都單獨來完成，而不是和排程算法一起，因為一起做考慮，畢竟很複雜，也許那樣性能會好一些（不需要一次一次的周遊所有資源池和任務清單）。在實作算法的時候，需要考慮時間和資源需求的雙重搭配，完成一次計算需要重新計算排程，已經計算資源可以重複利用，每次完成一次排程，需要重新更新資源池大小和任務隊列，很多情況需要考慮。每一種情況有需要完成一點就驗證一點，具體就不詳細解釋了。
3.最終完成介紹
（1）把實作的功能打成一個jar包，執行如下指令可以運作：
java
-jar taskdispatch-0.0.1-SNAPSHOT.jar
/home/wuyouqiang/workspace/taskdispatch/src/main/resource/task.yaml
/home/wuyouqiang/workspace/taskdispatch/src/main/resource/compute.conf
後面兩個參數分别是任務清單的配置檔案和計算資源池的配置檔案（具體配置檔案的路徑根據運作時的環境确定，上面的隻是我電腦上的示範）。如果沒有提供，程式有簡單提示，如果多了也隻認前兩個。
上面兩個配置檔案分别如下：
/home/wuyouqiang/workspace/taskdispatch/src/main/resource/task.yaml：
task1:
cores_required:
2
execution_time:
100
task2:
cores_required:
1
execution_time:
200
task3:
cores_required:
4
execution_time:
50
parent_tasks:
“task1, task2”
task4:
cores_required:
2
execution_time:
80
parent_tasks:
“task2”
/home/wuyouqiang/workspace/taskdispatch/src/main/resource/compute.conf：
compute1:
2
compute2:
2
compute3:
6
下面是運作的結果（中間有一些記錄任務排程過程，友善調試的資訊）：
開始排程任務 : [task1]到計算資源compute2
開始排程任務 : [task2]到計算資源compute1
删除任務
：task1
compute2恢複資源
： 2
删除任務[task3]的依賴任務 : [task1]
删除任務
：task2
compute1恢複資源
： 1
删除任務[task3]的依賴任務 : [task2]
删除任務[task4]的依賴任務 : [task2]
開始排程任務 : [task3]到計算資源compute3
開始排程任務 : [task4]到計算資源compute1
删除任務
：task3
compute3恢複資源
： 4
删除任務
：task4
compute1恢複資源
： 2
任務排程成功,排程結果如下:
task1:compute2
task2:compute1
task3:compute3
task4:compute1
（2）也提供maven或者eclipse功能，代碼裡有一些簡單的注釋，是自己寫代碼時的思考，不一定對，但對我的思路有一個記錄作用，裡面有一些簡單的測試樣例。
4.思考
（1）首先代碼肯定實作不夠完美，測試用例也不夠完善，最終實作的算法也不一定是最優化的，也不一定是完全正确的。隻能說在有限的時間裡完成了題目的要求，并且我也覺得可用了。
（2）這個隻是一個任務，如果真正做成一個完整的産品可能需要更多的考慮。我們目前完成這個最多算一個beta版本。其實産品的考慮更多提供一個通用的方式，例如提供接口直接傳遞任務和計算資源資訊。另外在網際網路公司本能的考慮到7*24小時的可靠性，是以怎麼做防單點，基本的利用zookeeper等從master和slave的切換等。
（3）想法太多，不過優先需要考慮怎麼在規定的時間裡把任務完成的最好，因為任何産品都是可以無止境的優化和改進。
（4）技術方案選擇上，同一個任務，有太多的方案可以選擇，而且沒有任何一種方案是絕對的有優勢，是以我一般選擇能夠滿足任務要求的并且最容易實作的。這個題目選擇了java的技術體系和eclipse+maven做為開發工具，雖然我對他們不是很熟悉，但是我相信通過網絡最容易找到他們的資料。是以選擇，如果選擇其他語言體系來實作，可能8-10小時不容易。
三，代碼實作
1.工程結構如下：
2.各個檔案代碼詳細實作，代碼中有注釋
（1）App.java
package com.fa.taskdispatch;
import java.util.ArrayList;
import java.util.HashMap;
public class App {
public static void main(String[] args) {
if (args.length < 2) {
System.out.println("Usage : 需要兩個運作參數，第一個是任務清單的檔案路徑，第二個是計算資源的檔案路徑");
System.exit(1);
}
AnalyticalResource ar = new AnalyticalResource(args[0], args[1]);
HashMap<String, Task> taskMap = new HashMap<String, Task>();
if (!ar.resolveTaskListFile(taskMap)) {
System.out.println("解析任務清單的YAML檔案失敗，系統退出.");
System.exit(1);
}
ArrayList<ComputeResource> computeResourceList = new ArrayList<ComputeResource>();
if (!ar.resolveComputeResourceFile(computeResourceList)) {
System.out.println("解析計算資源檔案失敗，系統退出.");
System.exit(1);
}
TaskDispatch td = new TaskDispatch(taskMap, computeResourceList);
ArrayList<String> dispatchResult = new ArrayList<String>();
if (td.dispatch(dispatchResult)) {
System.out.println("任務排程成功,排程結果如下: ");
for (String name : dispatchResult) {
System.out.println(name);
}
} else {
System.out.println("任務排程失敗.");
}
}
} 複制
(2) ComputeResource.java
package com.fa.taskdispatch;
public class ComputeResource {
private String name;
private int cores;
public ComputeResource(String name, int cores) {
this.name = name;
this.cores = cores;
}
public void subCores(int number) {
cores = cores - number;
}
public void addCores(int number) {
cores = cores + number;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public int getCores() {
return cores;
}
public void setCores(int cores) {
this.cores = cores;
}
} 複制
(3) Task.java
package com.fa.taskdispatch;
import java.util.ArrayList;
public class Task {
private String name;
private int coresRequired;
private int executionTime;
private ArrayList<String> parentTasks;
public Task(String name, int coresRequired, int executionTime) {
this(name, coresRequired, executionTime, new ArrayList<String>());
}
public Task(String name, int coresRequired, int executionTime, ArrayList<String> parentTasks) {
this.name = name;
this.coresRequired = coresRequired;
this.executionTime = executionTime;
this.parentTasks = parentTasks;
}
public boolean isDependMyself() {
boolean result = false;
for (String p : parentTasks) {
if (p.equals(name)) {
result = true;
}
}
return result;
}
public boolean parentTasksIsEmpty() {
return parentTasks == null || parentTasks.isEmpty();
}
//删除依賴任務中的一個
public void deleteOneParentTasks(String name) {
parentTasks.remove(name);
}
public void subExecutionTime(int number) {
executionTime = executionTime - number;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public int getCoresRequired() {
return coresRequired;
}
public void setCoresRequired(int coresRequired) {
this.coresRequired = coresRequired;
}
public int getExecutionTime() {
return executionTime;
}
public void setExecutionTime(int executionTime) {
this.executionTime = executionTime;
}
public ArrayList<String> getParentTasks() {
return parentTasks;
}
public void setParentTasks(ArrayList<String> parentTasks) {
this.parentTasks = parentTasks;
}
} 複制
(4) AnalyticalResource.java
package com.fa.taskdispatch;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import org.yaml.snakeyaml.Yaml;
public class AnalyticalResource {
private String taskListFile = "";
private String computeResource = "";
public AnalyticalResource(String taskListFile, String computeResource) {
this.taskListFile = taskListFile;
this.computeResource = computeResource;
}
public boolean resolveTaskListFile(HashMap<String, Task> tasks) {
boolean result = true;
if (tasks == null) {
tasks = new HashMap<String, Task>();
}
Yaml y = new Yaml();
File f = new File(taskListFile);
try {
HashMap tasksMap = (HashMap) y.load(new FileInputStream(f));
for (Object taskName : tasksMap.keySet()) {
HashMap taskMap = (HashMap)tasksMap.get(taskName);
int coresRequired = (Integer)taskMap.get("cores_required");
int executionTime = (Integer)taskMap.get("execution_time");
String parentTasks = (String)taskMap.get("parent_tasks");
String name = (String)taskName;
if (parentTasks != null && !parentTasks.isEmpty()) {
String parents[] = parentTasks.split(",");
ArrayList<String> parentList = new ArrayList<String>();
for (String parent : parents) {
parentList.add(parent.trim());
}
Task t = new Task(name.trim(), coresRequired, executionTime, parentList);
tasks.put(name.trim(), t);
} else {
Task t = new Task(name.trim(), coresRequired, executionTime);
tasks.put(name.trim(), t);
}
}
} catch (FileNotFoundException e) {
// TODO Auto-generated catch block
result = false;
e.printStackTrace();
}
return result;
}
public boolean resolveComputeResourceFile(ArrayList<ComputeResource> computeResourceList){
boolean result = true;
if (computeResourceList == null) {
computeResourceList = new ArrayList<ComputeResource>();
}
try {
Properties pt = new Properties();
pt.load(new FileInputStream(computeResource));
if (pt.isEmpty()) {
result = false;
System.out.println("沒有計算資源，沒有辦法進行排程，也算失敗，需要退出.");
}
for (Map.Entry<Object, Object> entry : pt.entrySet()) {
String name = (String)entry.getKey();
int cores = Integer.parseInt(entry.getValue().toString().trim());
ComputeResource cr = new ComputeResource(name.trim(), cores);
computeResourceList.add(cr);
}
} catch (FileNotFoundException e) {
// TODO Auto-generated catch block
result = false;
e.printStackTrace();
} catch (IOException e) {
// TODO Auto-generated catch block
result = false;
e.printStackTrace();
}
return result;
}
} 複制
(5)最重要的類TaskDispatch.java
package com.fa.taskdispatch;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
public class TaskDispatch {
private HashMap<String, Task> taskMap;
private ArrayList<ComputeResource> computeResourceList;
public TaskDispatch(HashMap<String, Task> taskMap, ArrayList<ComputeResource> computeResourceList) {
this.taskMap = taskMap;
this.computeResourceList = computeResourceList;
}
//檢查從某一個任務開始是否存在循環依賴
public boolean checkOneTaskCycleTaskDepend(Task t) {
//檢查循環依賴就是循環的檢查parent_tasks是否會到達任務本身
//這個是需要改變的，所有單獨copy一份出來
ArrayList<String> parentList = new ArrayList<String>(t.getParentTasks());
String checkTaskName = t.getName();
while (parentList != null && !parentList.isEmpty()) {
ArrayList<String> newParentList = new ArrayList<String>();
for (String parent : parentList) {
if (parent.equals(checkTaskName)) {
System.out.println("任務["+ checkTaskName + "]存在了循環依賴，不能完成排程.");
return true;
}
Task dependTask = taskMap.get(parent);
if (dependTask.isDependMyself()) {
System.out.println("任務["+ dependTask.getName() + "]存在自己依賴自己的循環依賴，不能完成排程.");
return true;
}
if (dependTask!= null && !dependTask.parentTasksIsEmpty()) {
newParentList.addAll(dependTask.getParentTasks());
}
}
parentList = newParentList;
}
return false;
}
//檢查任務是否存在循環依賴，如果存在也是不能排程
public boolean checkCycleTaskDepend() {
boolean result = false;
for (String taskName : taskMap.keySet()) {
Task t = taskMap.get(taskName);
if (t.parentTasksIsEmpty()) {
continue;
}
if (checkOneTaskCycleTaskDepend(t)) {
result = true;
}
if (result) {
break;
}
}
return result;
}
public boolean dispatch(ArrayList<String> dispatchResults) {
//1.驗證可排程，如果有一個任務需要的cores_required超過了所有計算機的資源就不能排程
//(1)找出所有任務中需要最大cores_required
boolean result = true;
int maxCoresRequired = 0;
for (String taskName : taskMap.keySet()) {
Task t = taskMap.get(taskName);
if (t.getCoresRequired() > maxCoresRequired) {
maxCoresRequired = t.getCoresRequired();
}
}
//(2)找出計算資源的最大值
int maxComputeResource = 0;
for (ComputeResource cr : computeResourceList) {
if (cr.getCores() > maxComputeResource) {
maxComputeResource = cr.getCores();
}
}
//(3)比較
if (maxCoresRequired > maxComputeResource) {
System.out.println("單個任務需要的計算資源大于了任何一台計算機能夠提供的資源，是以不能完成這個任務");
return false;
}
//2.驗證是否有循環依賴
if (checkCycleTaskDepend()) {
System.out.println("任務清單裡面存在這個循環依賴，不能完成任務排程");
return false;
}
//3.排程任務：首先需要做任務的編排（因為任務之間有依賴）
//（1）首先找出本次可以排程的所有任務（對于第一次當然就是沒有任何依賴的任務）；
//（2）然後有一個以上的任務完成以後在查找依賴這些以完成任務的任務是否可以進行排程了；
//（3）每次排程都需要有資源的情況下進行；
//（4）最終形成一個可調用的任務和可使用的計算資源進行比對；
if (dispatchResults == null) {
dispatchResults = new ArrayList<String>();
}
ArrayList<Task> readyTask = new ArrayList<Task>();
HashMap<String, Task> runningTask = new HashMap<String, Task>();
while (!taskMap.isEmpty()) {//直到所有任務都配置設定計算資源以後結束排程
for (String taskName : taskMap.keySet()) {
Task t = taskMap.get(taskName);
if (t.parentTasksIsEmpty() && !readyTask.contains(t)) {
readyTask.add(t);
}
}
//根據可排程的任務和可以使用的資源進行一次排程，排程之前先對任務和資源進行排序
//1.把可以排程的任務按照時間從短到長排序
Comparator<Task> taskComparator = new Comparator<Task>(){
public int compare(Task t1, Task t2) {
return t1.getExecutionTime() - t2.getExecutionTime();
}
};
Collections.sort(readyTask, taskComparator);
//2.把可以利用的計算資源排序
Comparator<ComputeResource> computeComparator = new Comparator<ComputeResource>(){
public int compare(ComputeResource c1, ComputeResource c2) {
return c1.getCores() - c2.getCores();
}
};
Collections.sort(computeResourceList, computeComparator);
oneDispatchTask(dispatchResults, runningTask, readyTask, computeResourceList);
//排程完成一次，需要把時間最短（可能有多個）的任務結束掉，從可排程的任務裡面删除掉，并且重新計算可排程的任務，并且恢複資源
int minTime = readyTask.get(0).getExecutionTime();
ArrayList<Task> deleteTaskName = new ArrayList<Task>();
for (Task t : readyTask) {
if (minTime >= t.getExecutionTime()) {//删除掉已經完成的任務，就是時間最小的
System.out.println("删除任務 ：" + t.getName());
deleteTaskName.add(t);//記住删除了那些任務，後面用于更新readyTask
runningTask.remove(t.getName());
taskMap.remove(t.getName());
//恢複資源
for (String name : dispatchResults) {
if (name.contains(t.getName())) {
String arr[] = name.split(":");
String computeName = arr[1];
for (ComputeResource cr : computeResourceList) {
if (cr.getName().equals(computeName)) {
System.out.println(computeName + "恢複資源 ： " + t.getCoresRequired());
cr.addCores(t.getCoresRequired());
}
}
}
}
//繼續删除其他任務依賴這個任務的依賴任務
for (String taskName : taskMap.keySet()) {
Task task = taskMap.get(taskName);
if (task.getParentTasks().contains(t.getName())) {
task.deleteOneParentTasks(t.getName());
System.out.println("删除任務[" + task.getName() + "]的依賴任務 : [" + t.getName() + "]" );
}
}
}
}
//其他已經在排程的任務需要減去這個時間
for (String taskName : runningTask.keySet()) {
Task t = runningTask.get(taskName);
t.subExecutionTime(minTime);
}
//删除可排程任務裡面的已經完成的任務
for (Task t : deleteTaskName) {
readyTask.remove(t);
}
}
return result;
}
//排程算法：每次排程的時候選擇按照從時間從短到長的排序，優先排程時間短的，并且找一個資源能夠滿足排程任務的最小的計算資源配置設定
public void oneDispatchTask(ArrayList<String> dispatchResult, HashMap<String, Task> runningTask,
ArrayList<Task> readyTask, ArrayList<ComputeResource> availableResources) {
for (Task t : readyTask) {
//如果任務已經在運作就不需要在繼續配置設定資源排程了
if (runningTask.containsKey(t.getName())) {
continue;
}
for (ComputeResource cr : availableResources) {
//如果滿足資源需求，就配置設定這個任務到這一個計算資源了
if (t.getCoresRequired() <= cr.getCores()) {
dispatchResult.add(t.getName() + ":" + cr.getName());
runningTask.put(t.getName(), t);
System.out.println("開始排程任務 : [" + t.getName() + "]到計算資源" + cr.getName());
//減去這個計算資源已經使用的cores
cr.subCores(t.getCoresRequired());
break;
}
}
}
}
} 複制