Multi-phase training service (#148)

* Dev enas - multi-phase hyper parameters support (#96) * Multi-phase support * Updates * Updates * updates * updates * updates * Merge master to dev-enas (#117) * Multi-phase support * update document (#92) * Edit readme.md * updated a word * Update GetStarted.md * Update GetStarted.md * refact readme, getstarted and write your trial md. * Update README.md * Update WriteYourTrial.md * Update WriteYourTrial.md * Update WriteYourTrial.md * Update WriteYourTrial.md * Fix nnictl bugs and add new feature (#75) * fix nnictl bug * fix nnictl create bug * add experiment status logic * add more information for nnictl * fix Evolution Tuner bug * refactor code * fix code in updater.py * fix nnictl --help * fix classArgs bug * update check response.status_code logic * Updates * remove Buffer warning (#100) * update readme in ga_squad * update readme * fix typo * Update README.md * Update README.md * Update README.md * Updates * updates * updates * updates * Add support for debugging mode * fix setup.py (#115) * Add DAG model configuration format for SQuAD example. * Explain config format for SQuAD QA model. * Add more detailed introduction about the evolution algorithm. * Merge master to dev-enas (#118) * update document (#92) * Edit readme.md * updated a word * Update GetStarted.md * Update GetStarted.md * refact readme, getstarted and write your trial md. * Update README.md * Update WriteYourTrial.md * Update WriteYourTrial.md * Update WriteYourTrial.md * Update WriteYourTrial.md * Fix nnictl bugs and add new feature (#75) * fix nnictl bug * fix nnictl create bug * add experiment status logic * add more information for nnictl * fix Evolution Tuner bug * refactor code * fix code in updater.py * fix nnictl --help * fix classArgs bug * update check response.status_code logic * remove Buffer warning (#100) * update readme in ga_squad * update readme * fix typo * Update README.md * Update README.md * Update README.md * Add support for debugging mode * fix setup.py (#115) * Add DAG model configuration format for SQuAD example. * Explain config format for SQuAD QA model. * Add more detailed introduction about the evolution algorithm. * Fix install.sh add add trial log path (#109) * fix nnictl bug * fix nnictl create bug * add experiment status logic * add more information for nnictl * fix Evolution Tuner bug * refactor code * fix code in updater.py * fix nnictl --help * fix classArgs bug * update check response.status_code logic * show trial log path * update document * fix install.sh * set default vallue for maxTrialNum and maxExecDuration * fix nnictl * support multiPhase (#127) * fix nnictl bug * support multiPhase * Fix multiphase datastore problem (#125) * Fix multiphase datastore problem * updates * updates * updates * updates * Pull latest code (#2) * webui logpath and document (#135) * Add webui document and logpath as a href * fix tslint * fix comments by Chengmin * Pai training service bug fix and enhancement (#136) * Add NNI installation scripts * Update pai script, update NNI_out_dir * Update NNI dir in nni sdk local.py * Create .nni folder in nni sdk local.py * Add check before creating .nni folder * Fix typo for PAI_INSTALL_NNI_SHELL_FORMAT * Improve annotation (#138) * Improve annotation * Minor bugfix * Selectively install through pip (#139) Selectively install through pip * update setup.py * fix paiTrainingService bugs (#137) * fix nnictl bug * add hdfs host validation * fix bugs * fix dockerfile * fix install.sh * update install.sh * fix dockerfile * Set timeout for HDFSUtility exists function * remove unused TODO * fix sdk * add optional for outputDir and dataDir * refactor dockerfile.base * Remove unused import in hdfsclientUtility * Add documentation for NNI PAI mode experiment (#141) * Add documentation for NNI PAI mode * Fix typo based on PR comments * Exit with subprocess return code of trial keeper * Remove additional exit code * Fix typo based on PR comments * update doc for smac tuner (#140) * Revert "Selectively install through pip (#139)" due to potential pip install issue (#142) * Revert "Selectively install through pip (#139)" This reverts commit 1d17483. * Add exit code of subprocess for trial_keeper * Update README, add link to PAImode doc * fix bug (#147) * Refactor nnictl and add config_pai.yml (#144) * fix nnictl bug * add hdfs host validation * fix bugs * fix dockerfile * fix install.sh * update install.sh * fix dockerfile * Set timeout for HDFSUtility exists function * remove unused TODO * fix sdk * add optional for outputDir and dataDir * refactor dockerfile.base * Remove unused import in hdfsclientUtility * add config_pai.yml * refactor nnictl create logic and add colorful print * fix nnictl stop logic * add annotation for config_pai.yml * add document for start experiment * fix config.yml * fix document * Fix trial keeper wrongly exit issue (#152) * Fix trial keeper bug, use actual exitcode to exit rather than 1 * Fix bug of table sort (#145) * Update doc for PAIMode and v0.2 release notes (#153) * Update v0.2 documentation regards to release note and PAI training service * Update document to describe NNI docker image * Bug fix for SQuAD example tuner. (#134) * Update Makefile (#151) * test * update setup.py * update Makefile and install.sh * rever setup.py * change color * update doc * update doc * fix auto-completion's extra space * update Makefile * update webui * Update doc image (#163) * update doc * trivial * trivial * trivial * trivial * trivial * trivial * update image * update image size * Update ga squad (#104) * update readme in ga_squad * update readme * fix typo * Update README.md * Update README.md * Update README.md * update readme * sklearn examples (#169) * fix nnictl bug * fix install.sh * add sklearn-regression example * add sklearn classification * update sklearn * update example * remove additional code * Update batch tuner (#158) * update readme in ga_squad * update readme * fix typo * Update README.md * Update README.md * Update README.md * update readme * update batch tuner * Quickly fix cascading search space bug in tuner (#156) * update readme in ga_squad * update readme * fix typo * Update README.md * Update README.md * Update README.md * update readme * quickly fix cascading searchspace bug in tuner * Add iterative search space example (#119) * update readme in ga_squad * update readme * fix typo * Update README.md * Update README.md * Update README.md * update readme * add iterative search space example * update * update readme * change name * updates * updates * Updates CI * updates
SparkSnail · Oct 8, 2018 · 3908578 · 3908578
1 parent 6ef6511
commit 3908578
Show file tree

Hide file tree

Showing 23 changed files with 559 additions and 47 deletions.
diff --git a/.travis.yml b/.travis.yml
@@ -11,7 +11,7 @@ before_install:
   - sudo sh -c 'PATH=/usr/local/node/bin:$PATH yarn global add serve'
 install:
   - make
-  - make install
+  - make easy-install
   - export PATH=$HOME/.nni/bin:$PATH
 before_script:
   - cd test/naive

diff --git a/src/nni_manager/common/datastore.ts b/src/nni_manager/common/datastore.ts
@@ -22,8 +22,8 @@
 import { ExperimentProfile, TrialJobStatistics } from './manager';
 import { TrialJobDetail, TrialJobStatus } from './trainingService';
 
-type TrialJobEvent = TrialJobStatus | 'USER_TO_CANCEL' | 'ADD_CUSTOMIZED';
-type MetricType = 'PERIODICAL' | 'FINAL' | 'CUSTOM';
+type TrialJobEvent = TrialJobStatus | 'USER_TO_CANCEL' | 'ADD_CUSTOMIZED' | 'ADD_HYPERPARAMETER';
+type MetricType = 'PERIODICAL' | 'FINAL' | 'CUSTOM' | 'REQUEST_PARAMETER';
 
 interface ExperimentProfileRecord {
     readonly timestamp: number;
@@ -62,7 +62,7 @@ interface TrialJobInfo {
     status: TrialJobStatus;
     startTime?: number;
     endTime?: number;
-    hyperParameters?: string;
+    hyperParameters?: string[];
     logPath?: string;
     finalMetricData?: string;
     stderrPath?: string;

diff --git a/src/nni_manager/common/manager.ts b/src/nni_manager/common/manager.ts
@@ -31,6 +31,7 @@ interface ExperimentParams {
     maxExecDuration: number; //seconds
     maxTrialNum: number;
     searchSpace: string;
+    multiPhase?: boolean;
     tuner: {
         className: string;
         builtinTunerName?: string;

diff --git a/src/nni_manager/common/trainingService.ts b/src/nni_manager/common/trainingService.ts
@@ -37,11 +37,16 @@ interface JobApplicationForm {
     readonly jobType: JobType;
 }
 
+interface HyperParameters {
+    readonly value: string;
+    readonly index: number;
+}
+
 /**
  * define TrialJobApplicationForm
  */
 interface TrialJobApplicationForm extends JobApplicationForm {
-    readonly hyperParameters: string;
+    readonly hyperParameters: HyperParameters;
 }
 
 /**
@@ -116,6 +121,6 @@ abstract class TrainingService {
 
 export {
     TrainingService, TrainingServiceError, TrialJobStatus, TrialJobApplicationForm,
-    TrainingServiceMetadata, TrialJobDetail, TrialJobMetric,
+    TrainingServiceMetadata, TrialJobDetail, TrialJobMetric, HyperParameters,
     HostJobApplicationForm, JobApplicationForm, JobType
 };
diff --git a/src/nni_manager/common/utils.ts b/src/nni_manager/common/utils.ts
@@ -158,8 +158,11 @@ function parseArg(names: string[]): string {
  * @param assessor: similiar as tuner
  *
  */
-function getMsgDispatcherCommand(tuner: any, assessor: any): string {
+function getMsgDispatcherCommand(tuner: any, assessor: any, multiPhase: boolean = false): string {
     let command: string = `python3 -m nni --tuner_class_name ${tuner.className}`;
+    if (multiPhase) {
+        command += ' --multi_phase';
+    }
 
     if (tuner.classArgs !== undefined) {
         command += ` --tuner_args ${JSON.stringify(JSON.stringify(tuner.classArgs))}`;

diff --git a/src/nni_manager/core/commands.ts b/src/nni_manager/core/commands.ts
@@ -27,6 +27,7 @@ const TRIAL_END = 'EN';
 const TERMINATE = 'TE';
 
 const NEW_TRIAL_JOB = 'TR';
+const SEND_TRIAL_JOB_PARAMETER = 'SP';
 const NO_MORE_TRIAL_JOBS = 'NO';
 const KILL_TRIAL_JOB = 'KI';
 
@@ -39,6 +40,7 @@ const TUNER_COMMANDS: Set<string> = new Set([
     TERMINATE,
 
     NEW_TRIAL_JOB,
+    SEND_TRIAL_JOB_PARAMETER,
     NO_MORE_TRIAL_JOBS
 ]);
 
@@ -63,5 +65,6 @@ export {
     NO_MORE_TRIAL_JOBS,
     KILL_TRIAL_JOB,
     TUNER_COMMANDS,
-    ASSESSOR_COMMANDS
+    ASSESSOR_COMMANDS,
+    SEND_TRIAL_JOB_PARAMETER
 };
diff --git a/src/nni_manager/core/nniDataStore.ts b/src/nni_manager/core/nniDataStore.ts
@@ -26,6 +26,7 @@ import * as component from '../common/component';
 import { Database, DataStore, MetricData, MetricDataRecord, MetricType,
     TrialJobEvent, TrialJobEventRecord, TrialJobInfo } from '../common/datastore';
 import { isNewExperiment } from '../common/experimentStartupInfo';
+import { getExperimentId } from '../common/experimentStartupInfo';
 import { getLogger, Logger } from '../common/log';
 import { ExperimentProfile,  TrialJobStatistics } from '../common/manager';
 import { TrialJobStatus } from '../common/trainingService';
@@ -35,6 +36,7 @@ class NNIDataStore implements DataStore {
     private db: Database = component.get(Database);
     private log: Logger = getLogger();
     private initTask!: Deferred<void>;
+    private multiPhase: boolean | undefined;
 
     public init(): Promise<void> {
         if (this.initTask !== undefined) {
@@ -112,13 +114,19 @@ class NNIDataStore implements DataStore {
     }
 
     public async getTrialJob(trialJobId: string): Promise<TrialJobInfo> {
-        const trialJobs = await this.queryTrialJobs(undefined, trialJobId);
+        const trialJobs: TrialJobInfo[] = await this.queryTrialJobs(undefined, trialJobId);
 
         return trialJobs[0];
     }
 
     public async storeMetricData(trialJobId: string, data: string): Promise<void> {
-        const metrics = JSON.parse(data) as MetricData;
+        const metrics: MetricData = JSON.parse(data);
+        // REQUEST_PARAMETER is used to request new parameters for multiphase trial job,
+        // it is not metrics, so it is skipped here.
+        if (metrics.type === 'REQUEST_PARAMETER') {
+
+            return;
+        }
         assert(trialJobId === metrics.trial_job_id);
         await this.db.storeMetricData(trialJobId, JSON.stringify({
             trialJobId: metrics.trial_job_id,
@@ -160,25 +168,56 @@ class NNIDataStore implements DataStore {
 
     private async getFinalMetricData(trialJobId: string): Promise<any> {
         const metrics: MetricDataRecord[] = await this.getMetricData(trialJobId, 'FINAL');
-        if (metrics.length > 1) {
-            this.log.error(`Found multiple final results for trial job: ${trialJobId}`);
+
+        const multiPhase: boolean = await this.isMultiPhase();
+
+        if (metrics.length > 1 && !multiPhase) {
+            this.log.error(`Found multiple FINAL results for trial job ${trialJobId}`);
+        }
+
+        return metrics[metrics.length - 1];
+    }
+
+    private async isMultiPhase(): Promise<boolean> {
+        if (this.multiPhase === undefined) {
+            this.multiPhase = (await this.getExperimentProfile(getExperimentId())).params.multiPhase;
         }
 
-        return metrics[0];
+        if (this.multiPhase !== undefined) {
+            return this.multiPhase;
+        } else {
+            return false;
+        }
     }
 
-    private getJobStatusByLatestEvent(event: TrialJobEvent): TrialJobStatus {
+    private getJobStatusByLatestEvent(oldStatus: TrialJobStatus, event: TrialJobEvent): TrialJobStatus {
         switch (event) {
             case 'USER_TO_CANCEL':
                 return 'USER_CANCELED';
             case 'ADD_CUSTOMIZED':
                 return 'WAITING';
+            case 'ADD_HYPERPARAMETER':
+                return oldStatus;
             default:
         }
 
         return <TrialJobStatus>event;
     }
 
+    private mergeHyperParameters(hyperParamList: string[], newParamStr: string): string[] {
+        const mergedHyperParams: any[] = [];
+        const newParam: any = JSON.parse(newParamStr);
+        for (const hyperParamStr of hyperParamList) {
+            const hyperParam: any = JSON.parse(hyperParamStr);
+            mergedHyperParams.push(hyperParam);
+        }
+        if (mergedHyperParams.filter((value: any) => value.parameter_index === newParam.parameter_index).length <= 0) {
+            mergedHyperParams.push(newParam);
+        }
+
+        return mergedHyperParams.map<string>((value: any) => { return JSON.stringify(value); });
+    }
+
     private getTrialJobsByReplayEvents(trialJobEvents: TrialJobEventRecord[]):  Map<string, TrialJobInfo> {
         const map: Map<string, TrialJobInfo> = new Map();
         // assume data is stored by time ASC order
@@ -192,7 +231,8 @@ class NNIDataStore implements DataStore {
             } else {
                 jobInfo = {
                     id: record.trialJobId,
-                    status: this.getJobStatusByLatestEvent(record.event)
+                    status: this.getJobStatusByLatestEvent('UNKNOWN', record.event),
+                    hyperParameters: []
                 };
             }
             if (!jobInfo) {
@@ -221,9 +261,13 @@ class NNIDataStore implements DataStore {
                     }
                 default:
             }
-            jobInfo.status = this.getJobStatusByLatestEvent(record.event);
+            jobInfo.status = this.getJobStatusByLatestEvent(jobInfo.status, record.event);
             if (record.data !== undefined && record.data.trim().length > 0) {
-                jobInfo.hyperParameters = record.data;
+                if (jobInfo.hyperParameters !== undefined) {
+                    jobInfo.hyperParameters = this.mergeHyperParameters(jobInfo.hyperParameters, record.data);
+                } else {
+                    assert(false, 'jobInfo.hyperParameters is undefined');
+                }
             }
             map.set(record.trialJobId, jobInfo);
         }

diff --git a/src/nni_manager/core/nnimanager.ts b/src/nni_manager/core/nnimanager.ts
@@ -37,7 +37,7 @@ import {
 import { delay , getLogDir, getMsgDispatcherCommand} from '../common/utils';
 import {
     ADD_CUSTOMIZED_TRIAL_JOB, KILL_TRIAL_JOB, NEW_TRIAL_JOB, NO_MORE_TRIAL_JOBS, REPORT_METRIC_DATA,
-    REQUEST_TRIAL_JOBS, TERMINATE, TRIAL_END, UPDATE_SEARCH_SPACE
+    REQUEST_TRIAL_JOBS, SEND_TRIAL_JOB_PARAMETER, TERMINATE, TRIAL_END, UPDATE_SEARCH_SPACE
 } from './commands';
 import { createDispatcherInterface, IpcInterface } from './ipcInterface';
 import { TrialJobMaintainerEvent, TrialJobs } from './trialJobs';
@@ -116,7 +116,7 @@ class NNIManager implements Manager {
         await this.storeExperimentProfile();
         this.log.debug('Setup tuner...');
 
-        const dispatcherCommand: string = getMsgDispatcherCommand(expParams.tuner, expParams.assessor);
+        const dispatcherCommand: string = getMsgDispatcherCommand(expParams.tuner, expParams.assessor, expParams.multiPhase);
         console.log(`dispatcher command: ${dispatcherCommand}`);
         this.setupTuner(
             //expParams.tuner.tunerCommand,
@@ -140,7 +140,7 @@ class NNIManager implements Manager {
         this.experimentProfile = await this.dataStore.getExperimentProfile(experimentId);
         const expParams: ExperimentParams = this.experimentProfile.params;
 
-        const dispatcherCommand: string = getMsgDispatcherCommand(expParams.tuner, expParams.assessor);
+        const dispatcherCommand: string = getMsgDispatcherCommand(expParams.tuner, expParams.assessor, expParams.multiPhase);
         console.log(`dispatcher command: ${dispatcherCommand}`);
         this.setupTuner(
             dispatcherCommand,
@@ -462,7 +462,10 @@ class NNIManager implements Manager {
                     this.currSubmittedTrialNum++;
                     const trialJobAppForm: TrialJobApplicationForm = {
                         jobType: 'TRIAL',
-                        hyperParameters: content
+                        hyperParameters: {
+                            value: content,
+                            index: 0
+                        }
                     };
                     const trialJobDetail: TrialJobDetail = await this.trainingService.submitTrialJob(trialJobAppForm);
                     this.trialJobsMaintainer.setTrialJob(trialJobDetail.id, Object.assign({}, trialJobDetail));
@@ -474,6 +477,22 @@ class NNIManager implements Manager {
                     }
                 }
                 break;
+            case SEND_TRIAL_JOB_PARAMETER:
+                const tunerCommand: any = JSON.parse(content);
+                assert(tunerCommand.parameter_index >= 0);
+                assert(tunerCommand.trial_job_id !== undefined);
+
+                const trialJobForm: TrialJobApplicationForm = {
+                    jobType: 'TRIAL',
+                    hyperParameters: {
+                        value: content,
+                        index: tunerCommand.parameter_index
+                    }
+                };
+                await this.trainingService.updateTrialJob(tunerCommand.trial_job_id, trialJobForm);
+                await this.dataStore.storeTrialJobEvent(
+                        'ADD_HYPERPARAMETER', tunerCommand.trial_job_id, content, undefined);
+                break;
             case NO_MORE_TRIAL_JOBS:
                 this.trialJobsMaintainer.setNoMoreTrials();
                 break;

diff --git a/src/nni_manager/rest_server/restValidationSchemas.ts b/src/nni_manager/rest_server/restValidationSchemas.ts
@@ -57,6 +57,7 @@ export namespace ValidationSchemas {
             trialConcurrency: joi.number().min(0).required(),
             searchSpace: joi.string().required(),
             maxExecDuration: joi.number().min(0).required(),
+            multiPhase: joi.boolean(),
             tuner: joi.object({
                 builtinTunerName: joi.string().valid('TPE', 'Random', 'Anneal', 'Evolution', 'SMAC', 'BatchTuner'),
                 codeDir: joi.string(),

diff --git a/src/nni_manager/training_service/local/localTrainingService.ts b/src/nni_manager/training_service/local/localTrainingService.ts
@@ -30,10 +30,11 @@ import { getLogger, Logger } from '../../common/log';
 import { TrialConfig } from '../common/trialConfig';
 import { TrialConfigMetadataKey } from '../common/trialConfigMetadataKey';
 import {
-    HostJobApplicationForm, JobApplicationForm, TrainingService, TrialJobApplicationForm,
+    HostJobApplicationForm, JobApplicationForm, HyperParameters, TrainingService, TrialJobApplicationForm,
     TrialJobDetail, TrialJobMetric, TrialJobStatus
 } from '../../common/trainingService';
 import { delay, getExperimentRootDir, uniqueString } from '../../common/utils';
+import { file } from 'tmp';
 
 const tkill = require('tree-kill');
 
@@ -210,8 +211,18 @@ class LocalTrainingService implements TrainingService {
      * @param trialJobId trial job id
      * @param form job application form
      */
-    public updateTrialJob(trialJobId: string, form: JobApplicationForm): Promise<TrialJobDetail> {
-        throw new MethodNotImplementedError();
+    public async updateTrialJob(trialJobId: string, form: JobApplicationForm): Promise<TrialJobDetail> {
+        const trialJobDetail: undefined | TrialJobDetail = this.jobMap.get(trialJobId);
+        if (trialJobDetail === undefined) {
+            throw new Error(`updateTrialJob failed: ${trialJobId} not found`);
+        }
+        if (form.jobType === 'TRIAL') {
+            await this.writeParameterFile(trialJobDetail.workingDirectory, (<TrialJobApplicationForm>form).hyperParameters);
+        } else {
+            throw new Error(`updateTrialJob failed: jobType ${form.jobType} not supported.`);
+        }
+
+        return trialJobDetail;
     }
 
     /**
@@ -332,10 +343,7 @@ class LocalTrainingService implements TrainingService {
         await cpp.exec(`mkdir -p ${path.join(trialJobDetail.workingDirectory, '.nni')}`);
         await cpp.exec(`touch ${path.join(trialJobDetail.workingDirectory, '.nni', 'metrics')}`);
         await fs.promises.writeFile(path.join(trialJobDetail.workingDirectory, 'run.sh'), runScriptLines.join('\n'), { encoding: 'utf8' });
-        await fs.promises.writeFile(
-            path.join(trialJobDetail.workingDirectory, 'parameter.cfg'),
-            (<TrialJobApplicationForm>trialJobDetail.form).hyperParameters,
-            { encoding: 'utf8' });
+        await this.writeParameterFile(trialJobDetail.workingDirectory, (<TrialJobApplicationForm>trialJobDetail.form).hyperParameters);
         const process: cp.ChildProcess = cp.exec(`bash ${path.join(trialJobDetail.workingDirectory, 'run.sh')}`);
 
         this.setTrialJobStatus(trialJobDetail, 'RUNNING');
@@ -402,6 +410,11 @@ class LocalTrainingService implements TrainingService {
             }
         }
     }
+
+    private async writeParameterFile(directory: string, hyperParameters: HyperParameters): Promise<void> {
+        const filepath: string = path.join(directory, `parameter_${hyperParameters.index}.cfg`);
+        await fs.promises.writeFile(filepath, hyperParameters.value, { encoding: 'utf8' });
+    }
 }
 
 export { LocalTrainingService };