Move the call of experimentDoneCleanUp into stopExperiment() method (#390)

* Adjust sleep position for sdk_test.py

* Exit dispather process if receive Terminate command

* Add comment for sleep change in sdk_test.py

src/nni_manager/core/nnimanager.ts

@@ -221,6 +221,9 @@ class NNIManager implements Manager {
 
     public async stopExperiment(): Promise<void> {
         this.status.status = 'STOPPING';
+        this.log.info('Experiment done, cleaning up...');
+        await this.experimentDoneCleanUp();
+        this.log.info('Experiment done.');
     }
 
     public async getMetricData(trialJobId?: string, metricType?: MetricType): Promise<MetricDataRecord[]> {
@@ -481,10 +484,6 @@ class NNIManager implements Manager {
             }
             await delay(1000 * 5); // 5 seconds
         }
-
-        this.log.info('Experiment done, cleaning up...');
-        await this.experimentDoneCleanUp();
-        this.log.info('Experiment done.');
     }
 
     private storeExperimentProfile(): Promise<void> {

src/nni_manager/training_service/kubeflow/kubeflowTrainingService.ts

@@ -206,7 +206,15 @@ class KubeflowTrainingService implements TrainingService {
     }
 
     public listTrialJobs(): Promise<TrialJobDetail[]> {
-        throw new MethodNotImplementedError();
+        const jobs: TrialJobDetail[] = [];
+        
+        this.trialJobsMap.forEach(async (value: KubeflowTrialJobDetail, key: string) => {
+            if (value.form.jobType === 'TRIAL') {
+                jobs.push(await this.getTrialJob(key));
+            }
+        });
+
+        return Promise.resolve(jobs);
     }
 
     public getTrialJob(trialJobId: string): Promise<TrialJobDetail> {
@@ -302,7 +310,7 @@ class KubeflowTrainingService implements TrainingService {
     }
 
     public getClusterMetadata(key: string): Promise<string> {
-        throw new MethodNotImplementedError();
+        return Promise.resolve('');
     }
 
     public async cleanUp(): Promise<void> {

src/sdk/pynni/nni/msg_dispatcher_base.py

@@ -65,7 +65,9 @@ class MsgDispatcherBase(Recoverable):
         _logger.debug('handle request: command: [{}], data: [{}]'.format(command, data))
 
         if command is CommandType.Terminate:
-            return False
+            # if receive Terminate command, exit process
+            _logger.info('Receive Terminate command from NNI manager, terminating')
+            exit(0)
 
         data = json_tricks.loads(data)
 

test/sdk_test.py

@@ -76,6 +76,8 @@ def run(dispatch_type):
     dipsatcher_list = TUNER_LIST if dispatch_type == 'Tuner' else ASSESSOR_LIST
     for dispatcher_name in dipsatcher_list:
         try:
+            # sleep 5 seconds here, to make sure previous stopped exp has enough time to exit to avoid port conflict
+            time.sleep(5)
             test_builtin_dispatcher(dispatch_type, dispatcher_name)
             print(GREEN + 'Test %s %s: TEST PASS' % (dispatcher_name, dispatch_type) + CLEAR)
         except Exception as error:
@@ -86,7 +88,6 @@ def run(dispatch_type):
         finally:
             subprocess.run(['nnictl', 'stop'])
 
-
 if __name__ == '__main__':
     installed = (sys.argv[-1] != '--preinstall')
     setup_experiment(installed)