Detect tuner failing (#635)

src/nni_manager/core/commands.ts

@@ -25,6 +25,7 @@ const UPDATE_SEARCH_SPACE = 'SS';
 const ADD_CUSTOMIZED_TRIAL_JOB = 'AD';
 const TRIAL_END = 'EN';
 const TERMINATE = 'TE';
+const PING = 'PI';
 
 const INITIALIZED = 'ID';
 const NEW_TRIAL_JOB = 'TR';
@@ -39,6 +40,7 @@ const TUNER_COMMANDS: Set<string> = new Set([
     UPDATE_SEARCH_SPACE,
     ADD_CUSTOMIZED_TRIAL_JOB,
     TERMINATE,
+    PING,
 
     INITIALIZED,
     NEW_TRIAL_JOB,
@@ -63,6 +65,7 @@ export {
     ADD_CUSTOMIZED_TRIAL_JOB,
     TRIAL_END,
     TERMINATE,
+    PING,
     INITIALIZED,
     NEW_TRIAL_JOB,
     NO_MORE_TRIAL_JOBS,

src/nni_manager/core/nnimanager.ts

@@ -35,15 +35,15 @@ import {
 import {
     TrainingService, TrialJobApplicationForm, TrialJobDetail, TrialJobMetric, TrialJobStatus
 } from '../common/trainingService';
-import { delay, getLogDir, getCheckpointDir, getMsgDispatcherCommand, mkDirP } from '../common/utils';
+import { delay, getCheckpointDir, getLogDir, getMsgDispatcherCommand, mkDirP } from '../common/utils';
 import {
-    ADD_CUSTOMIZED_TRIAL_JOB, INITIALIZE, INITIALIZED, KILL_TRIAL_JOB, NEW_TRIAL_JOB, NO_MORE_TRIAL_JOBS,
+    ADD_CUSTOMIZED_TRIAL_JOB, INITIALIZE, INITIALIZED, KILL_TRIAL_JOB, NEW_TRIAL_JOB, NO_MORE_TRIAL_JOBS, PING,
     REPORT_METRIC_DATA, REQUEST_TRIAL_JOBS, SEND_TRIAL_JOB_PARAMETER, TERMINATE, TRIAL_END, UPDATE_SEARCH_SPACE
 } from './commands';
 import { createDispatcherInterface, IpcInterface } from './ipcInterface';
 
 /**
- * NNIManager
+ * NNIManager which implements Manager interface
  */
 class NNIManager implements Manager {
     private trainingService: TrainingService;
@@ -360,6 +360,16 @@ class NNIManager implements Manager {
         }
     }
 
+    private async pingDispatcher(): Promise<void> {
+        if (this.dispatcher === undefined) {
+            throw new Error('Error: tuner has not been setup');
+        }
+        while (!['ERROR', 'STOPPING', 'STOPPED'].includes(this.status.status)) {
+            await delay(1000 * 5);
+            this.dispatcher.sendCommand(PING);
+        }
+    }
+
     private async requestTrialJobsStatus(): Promise<number> {
         let finishedTrialJobNum: number = 0;
         if (this.dispatcher === undefined) {
@@ -536,6 +546,9 @@ class NNIManager implements Manager {
 
         await Promise.all([
             this.periodicallyUpdateExecDuration(),
+            this.pingDispatcher().catch((err: Error) => {
+                throw new NNIError('Dispatcher error', `Dispatcher error: ${err.message}`, err);
+            }),
             this.trainingService.run().catch((err: Error) => {
                 throw new NNIError('Training service error', `Training service error: ${err.message}`, err);
             }),

src/sdk/pynni/nni/msg_dispatcher_base.py

@@ -83,7 +83,8 @@ class MsgDispatcherBase(Recoverable):
 
         _logger.debug('handle request: command: [{}], data: [{}]'.format(command, data))
 
-        data = json_tricks.loads(data)
+        if data:
+            data = json_tricks.loads(data)
 
         command_handlers = {
             # Tunner commands:
@@ -96,12 +97,16 @@ class MsgDispatcherBase(Recoverable):
             CommandType.ReportMetricData: self.handle_report_metric_data,
 
             CommandType.TrialEnd: self.handle_trial_end,
+            CommandType.Ping: self.handle_ping,
         }
         if command not in command_handlers:
             raise AssertionError('Unsupported command: {}'.format(command))
 
         return command_handlers[command](data)
 
+    def handle_ping(self, data):
+        pass
+
     def handle_initialize(self, data):
         raise NotImplementedError('handle_initialize not implemented')
 

src/sdk/pynni/nni/protocol.py

@@ -33,6 +33,7 @@ class CommandType(Enum):
     AddCustomizedTrialJob = b'AD'
     TrialEnd = b'EN'
     Terminate = b'TE'
+    Ping = b'PI'
 
     # out
     Initialized = b'ID'