一開始寫分散式處理程式的時候, 常常覺得會把 code 弄得很亂
因為要連線到 job-manager, 要指定 job 和裡面的 tasks
工作丟進去算, 算完還不一定對, 有錯也不會自動重算
整體來講, Matlab提供的 Distributed Computing Toolbox
直接要拿來用, 真的不怎麼好用 ^^"
經過一陣子嘗試, 我把要丟給 Distributed Computing 的任務
用一個 wrapper (包裝器) 包裝起來, 上面第一段提到的細節全部都寫在 wrapper
程式碼也變得比較精簡
wrapper的內容, 對於需要跟 Dist-Computing toolbox 打交道的朋友
應該也是有幫助
----
要丟到 wrapper 的程式, 必須寫成特定的形式
一般有三個步驟.
1. 準備傳給 function 的變數
2. 呼叫 wrapper 計算
3. 處理計算結果
首先, 假如我要運算的 function 是 ret = fun(a, b);
那麼 fun() 有兩個輸入和一個輸出, 假設我希望計算下列的結果...
ret(1) = fun(1, 10);
ret(2) = fun(2, 10);
...
ret(10) = fun(10, 10);
----
step.1 準備傳給 function 的變數
把要傳入 fun() 的傳入值用 cell-array 包裝起來
input = { {[1] [10]}, {[2] [10]}, ... {10, 10} };
* input 本身是一個大的 cell array
* input 裡面有很多個小的 cell-array, eg. { [1] [10] }
* 小 cell-array 裡, 每個元素, 分別對應到 function 的輸入參數
* 以 { [1] [10] } 來說, 到時候會算 fun([1], [10])
概念上, 先把所有的 input 全部都準備好
用 Matlab 的術語來說, input 變數在描述 "job"
input 變數裡每個小 cell, 在描述每個 "task"
step2. 呼叫 wrapper 計算
fhandle = @fun; % 告訴 wrapper 要算的 function-name
num_of_outputs = 1; % 告訴 wrapper 函數輸出的數目
is_destroy = 1; % 要不要清除在 jobmanager 上的運算結果
ret = dist_computing_wrapper_jobs( ...
fhandle, num_of_outputs, input, is_destroy);
step3. 處理計算結果
下面舉例, 把10次算完的結果畫成 10 張圖, 其他 case 依樣畫葫蘆即可
for ret_index=1:10
figure; plot( ret{ret_index} );
end
----
已知問題.
1. 這個 wrapper 會自動把算錯的 task 重新執行, 如果原本程式就寫錯
執行過程中本來就會錯誤, 那套用這個 wrapper 會引發無窮迴圈
2. 下面標示藍色的程式區塊, 請按照需要設定
----
function ret = dist_computing_wrapper_jobs(fhandle, num_of_outputs, input, is_destroy)
%% 連線到 job manager
jobmanager_name = 'jobmanager_xeon';
jobmanager_host = 'xeon';
jm = findResource( ...
'scheduler', 'type', 'jobmanager', ...
'name',jobmanager_name, ...
'LookupURL',jobmanager_host);
%% 建立 jobs 和 tasks
[len_job len_task] = size(input);
jobs = [];
for job_index=1:len_job
% 準備 jobs 並且丟去給他算
%% 建立 job 和裡面的 task
j = createJob(jm);
% 描述 job 裡的 tasks
for index_task = 1:len_task
createTask(j, fhandle, num_of_outputs, input(job_index, index_task));
end
% 設定檔案分享
set(j, 'FileDependencies', {'.' '../delsig'});
set(j, 'PathDependencies', {'.' '../delsig'});
% 設定 worker 在第一次執行 job 裡的 task 時, reset
set(j, 'RestartWorker', true);
% DEBUG 看 錯誤訊息
alltasks = get(j, 'Tasks');
set(alltasks, 'CaptureCommandWindowOutput', true);
% 提交 job 並執行
submit(j);
jobs = [jobs; j];
fprintf('Job #%d submitted\n', job_index);
end
%% 等所有jobs執行結束 & 取回執行結果
ret = [];
for job_index=1:len_job
waitForState(jobs(job_index));
% 檢查是不是所有 job 都正確
tasks = findTask(jobs(job_index));
error_task = [];
for task_index=1:length(tasks)
if(isempty(tasks(task_index).ErrorIdentifier) == 0) % Error task
error_task = [error_task task_index];
end
end
ret_currjob = getAllOutputArguments(jobs(job_index)); % 取回執行結果
if(isempty(error_task) == 0) % 有錯誤的結果
fprintf('執行有碰到錯誤, 自動重算\n');
input_redo = input(error_task);
ret_currjob_redo = dist_computing_wrapper_jobs(fhandle, num_of_outputs, input_redo, is_destroy);
for error_task_index=1:length(error_task)
ret_currjob(error_task(error_task_index)) = ret_currjob_redo(error_task_index);
end
end
fprintf('Job #%d completed ^+++^\n', job_index);
ret = [ret; ret_currjob];
end
if(is_destroy)
destroy(jobs)
end
--
※ 發信站: 批踢踢實業坊(ptt.cc)
◆ From: 140.113.236.3
※ 編輯: lihgong 來自: 140.113.236.3 (07/14 10:57)