Create experiment from Python code (#3111)

nni/runtime/protocol.py

@@ -6,6 +6,8 @@ import os
 import threading
 from enum import Enum
 
+_logger = logging.getLogger(__name__)
+
 
 class CommandType(Enum):
     # in
@@ -32,8 +34,7 @@ try:
         _in_file = open(3, 'rb')
         _out_file = open(4, 'wb')
 except OSError:
-    _msg = 'IPC pipeline not exists, maybe you are importing tuner/assessor from trial code?'
-    logging.getLogger(__name__).warning(_msg)
+    _logger.debug('IPC pipeline not exists')
 
 
 def send(command, data):
@@ -46,7 +47,7 @@ def send(command, data):
         _lock.acquire()
         data = data.encode('utf8')
         msg = b'%b%014d%b' % (command.value, len(data), data)
-        logging.getLogger(__name__).debug('Sending command, data: [%s]', msg)
+        _logger.debug('Sending command, data: [%s]', msg)
         _out_file.write(msg)
         _out_file.flush()
     finally:
@@ -58,14 +59,14 @@ def receive():
     Returns a tuple of command (CommandType) and payload (str)
     """
     header = _in_file.read(16)
-    logging.getLogger(__name__).debug('Received command, header: [%s]', header)
+    _logger.debug('Received command, header: [%s]', header)
     if header is None or len(header) < 16:
         # Pipe EOF encountered
-        logging.getLogger(__name__).debug('Pipe EOF encountered')
+        _logger.debug('Pipe EOF encountered')
         return None, None
     length = int(header[2:])
     data = _in_file.read(length)
     command = CommandType(header[:2])
     data = data.decode('utf8')
-    logging.getLogger(__name__).debug('Received command, data: [%s]', data)
+    _logger.debug('Received command, data: [%s]', data)
     return command, data

nni/utils.py

 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT license.
 
-import os
 import copy
 import functools
 from enum import Enum, unique
@@ -9,8 +8,6 @@ import json_tricks
 from schema import And
 
 from . import parameter_expressions
-from .runtime.common import init_logger
-from .runtime.env_vars import dispatcher_env_vars
 
 
 to_json = functools.partial(json_tricks.dumps, allow_nan=True)
@@ -120,16 +117,6 @@ def convert_dict2tuple(value):
     return value
 
 
-def init_dispatcher_logger():
-    """
-    Initialize dispatcher logging configuration
-    """
-    logger_file_path = 'dispatcher.log'
-    if dispatcher_env_vars.NNI_LOG_DIRECTORY is not None:
-        logger_file_path = os.path.join(dispatcher_env_vars.NNI_LOG_DIRECTORY, logger_file_path)
-    init_logger(logger_file_path, dispatcher_env_vars.NNI_LOG_LEVEL)
-
-
 def json2space(x, oldy=None, name=NodeType.ROOT):
     """
     Change search space from json format to hyperopt format

pylintrc

@@ -47,4 +47,4 @@ ignore-patterns=test*
 # List of members which are set dynamically and missed by pylint inference
 generated-members=numpy.*,torch.*,tensorflow.*
 
-ignored-modules=tensorflow
+ignored-modules=tensorflow,_winapi,msvcrt

setup.py

@@ -74,6 +74,7 @@ dependencies = [
     'websockets',
     'filelock',
     'prettytable',
+    'dataclasses ; python_version < "3.7"',
     'numpy < 1.19.4 ; sys_platform == "win32"',
     'numpy < 1.20 ; sys_platform != "win32" and python_version < "3.7"',
     'numpy ; sys.platform != "win32" and python_version >= "3.7"'

test/config/integration_tests.yml

@@ -143,8 +143,8 @@ testCases:
   config:
     maxTrialNum: 4
     trialConcurrency: 4
-  launchCommand: python3 -c 'from nni.experiment import Experiment; exp = Experiment(); exp.start_experiment("$configFile")'
-  stopCommand: python3 -c 'from nni.experiment import Experiment; exp = Experiment(); exp.connect_experiment("http://localhost:8080/"); exp.stop_experiment()'
+  launchCommand: python3 -c 'from nni.experiment import ExternalExperiment as Experiment; exp = Experiment(); exp.start_experiment("$configFile")'
+  stopCommand: python3 -c 'from nni.experiment import ExternalExperiment as Experiment; exp = Experiment(); exp.connect_experiment("http://localhost:8080/"); exp.stop_experiment()'
   validator:
     class: NnicliValidator
   platform: linux darwin

test/config/integration_tests_tf2.yml

@@ -110,8 +110,8 @@ testCases:
   config:
     maxTrialNum: 4
     trialConcurrency: 4
-  launchCommand: python3 -c 'from nni.experiment import Experiment; exp = Experiment(); exp.start_experiment("$configFile")'
-  stopCommand: python3 -c 'from nni.experiment import Experiment; exp = Experiment(); exp.connect_experiment("http://localhost:8080/"); exp.stop_experiment()'
+  launchCommand: python3 -c 'from nni.experiment import ExternalExperiment as Experiment; exp = Experiment(); exp.start_experiment("$configFile")'
+  stopCommand: python3 -c 'from nni.experiment import ExternalExperiment as Experiment; exp = Experiment(); exp.connect_experiment("http://localhost:8080/"); exp.stop_experiment()'
   validator:
     class: NnicliValidator
   platform: linux darwin

test/config/pr_tests.yml

@@ -47,8 +47,8 @@ testCases:
   config:
     maxTrialNum: 4
     trialConcurrency: 4
-  launchCommand: python3 -c 'from nni.experiment import Experiment; exp = Experiment(); exp.start_experiment("$configFile")'
-  stopCommand: python3 -c 'from nni.experiment import Experiment; exp = Experiment(); exp.connect_experiment("http://localhost:8080/"); exp.stop_experiment()'
+  launchCommand: python3 -c 'from nni.experiment import ExternalExperiment as Experiment; exp = Experiment(); exp.start_experiment("$configFile")'
+  stopCommand: python3 -c 'from nni.experiment import ExternalExperiment as Experiment; exp = Experiment(); exp.connect_experiment("http://localhost:8080/"); exp.stop_experiment()'
   validator:
     class: NnicliValidator
   platform: linux darwin

test/nni_test/nnitest/validators.py

@@ -6,7 +6,7 @@ from os import remove
 import subprocess
 import json
 import requests
-from nni.experiment import Experiment
+from nni.experiment import ExternalExperiment as Experiment
 from nni.tools.nnictl.updater import load_search_space
 from utils import METRICS_URL, GET_IMPORTED_DATA_URL
 

ts/nni_manager/common/experimentStartupInfo.ts

@@ -17,9 +17,10 @@ class ExperimentStartupInfo {
     private logDir: string = '';
     private logLevel: string = '';
     private readonly: boolean = false;
+    private dispatcherPipe: string | null = null;
     private platform: string = '';
 
-    public setStartupInfo(newExperiment: boolean, experimentId: string, basePort: number, platform: string, logDir?: string, logLevel?: string, readonly?: boolean): void {
+    public setStartupInfo(newExperiment: boolean, experimentId: string, basePort: number, platform: string, logDir?: string, logLevel?: string, readonly?: boolean, dispatcherPipe?: string): void {
         assert(!this.initialized);
         assert(experimentId.trim().length > 0);
         this.newExperiment = newExperiment;
@@ -41,6 +42,10 @@ class ExperimentStartupInfo {
         if (readonly !== undefined) {
             this.readonly = readonly;
         }
+
+        if (dispatcherPipe != undefined && dispatcherPipe.length > 0) {
+            this.dispatcherPipe = dispatcherPipe;
+        }
     }
 
     public getExperimentId(): string {
@@ -84,6 +89,11 @@ class ExperimentStartupInfo {
 
         return this.readonly;
     }
+
+    public getDispatcherPipe(): string | null {
+        assert(this.initialized);
+        return this.dispatcherPipe;
+    }
 }
 
 function getExperimentId(): string {
@@ -107,16 +117,20 @@ function getExperimentStartupInfo(): ExperimentStartupInfo {
 }
 
 function setExperimentStartupInfo(
-    newExperiment: boolean, experimentId: string, basePort: number, platform: string, logDir?: string, logLevel?: string, readonly?: boolean): void {
+    newExperiment: boolean, experimentId: string, basePort: number, platform: string, logDir?: string, logLevel?: string, readonly?: boolean, dispatcherPipe?: string): void {
     component.get<ExperimentStartupInfo>(ExperimentStartupInfo)
-        .setStartupInfo(newExperiment, experimentId, basePort, platform, logDir, logLevel, readonly);
+        .setStartupInfo(newExperiment, experimentId, basePort, platform, logDir, logLevel, readonly, dispatcherPipe);
 }
 
 function isReadonly(): boolean {
     return component.get<ExperimentStartupInfo>(ExperimentStartupInfo).isReadonly();
 }
 
+function getDispatcherPipe(): string | null {
+    return component.get<ExperimentStartupInfo>(ExperimentStartupInfo).getDispatcherPipe();
+}
+
 export {
     ExperimentStartupInfo, getBasePort, getExperimentId, isNewExperiment, getPlatform, getExperimentStartupInfo,
-    setExperimentStartupInfo, isReadonly
+    setExperimentStartupInfo, isReadonly, getDispatcherPipe
 };

ts/nni_manager/common/log.ts

@@ -126,7 +126,10 @@ class Logger {
      */
     private log(level: string, param: any[]): void {
         if (!this.readonly) {
-            const logContent = `[${(new Date()).toLocaleString()}] ${level} ${format(param)}\n`;
+            const time = new Date();
+            const localTime = new Date(time.getTime() - time.getTimezoneOffset() * 60000);
+            const timeStr = localTime.toISOString().slice(0, -5).replace('T', ' ');
+            const logContent = `[${timeStr}] ${level} ${format(param)}\n`;
             if (this.writable && this.bufferSerialEmitter) {
                 const buffer: WritableStreamBuffer = new WritableStreamBuffer();
                 buffer.write(logContent);

ts/nni_manager/core/ipcInterface.ts

@@ -6,6 +6,7 @@
 import * as assert from 'assert';
 import { ChildProcess } from 'child_process';
 import { EventEmitter } from 'events';
+import * as net from 'net';
 import { Readable, Writable } from 'stream';
 import { NNIError } from '../common/errors';
 import { getLogger, Logger } from '../common/log';
@@ -62,10 +63,10 @@ class IpcInterface {
      * @param proc the process to wrap
      * @param acceptCommandTypes set of accepted commands for this process
      */
-    constructor(proc: ChildProcess, acceptCommandTypes: Set<string>) {
+    constructor(outStream: Writable, inStream: Readable, acceptCommandTypes: Set<string>) {
         this.acceptCommandTypes = acceptCommandTypes;
-        this.outgoingStream = <Writable>proc.stdio[ipcOutgoingFd];
-        this.incomingStream = <Readable>proc.stdio[ipcIncomingFd];
+        this.outgoingStream = outStream;
+        this.incomingStream = inStream;
         this.eventEmitter = new EventEmitter();
         this.readBuffer = Buffer.alloc(0);
 
@@ -132,7 +133,14 @@ class IpcInterface {
  * @param process_ the tuner process
  */
 function createDispatcherInterface(process: ChildProcess): IpcInterface {
-    return new IpcInterface(process, new Set([...CommandType.TUNER_COMMANDS, ...CommandType.ASSESSOR_COMMANDS]));
+    const outStream = <Writable>process.stdio[ipcOutgoingFd];
+    const inStream = <Readable>process.stdio[ipcIncomingFd];
+    return new IpcInterface(outStream, inStream, new Set([...CommandType.TUNER_COMMANDS, ...CommandType.ASSESSOR_COMMANDS]));
 }
 
-export { IpcInterface, createDispatcherInterface, encodeCommand, decodeCommand };
+function createDispatcherPipeInterface(pipePath: string): IpcInterface {
+    const client = net.createConnection(pipePath);
+    return new IpcInterface(client, client, new Set([...CommandType.TUNER_COMMANDS, ...CommandType.ASSESSOR_COMMANDS]));
+}
+
+export { IpcInterface, createDispatcherInterface, createDispatcherPipeInterface, encodeCommand, decodeCommand };

ts/nni_manager/core/nnimanager.ts

@@ -9,7 +9,7 @@ import { Deferred } from 'ts-deferred';
 import * as component from '../common/component';
 import { DataStore, MetricDataRecord, MetricType, TrialJobInfo } from '../common/datastore';
 import { NNIError } from '../common/errors';
-import { getExperimentId } from '../common/experimentStartupInfo';
+import { getExperimentId, getDispatcherPipe } from '../common/experimentStartupInfo';
 import { getLogger, Logger } from '../common/log';
 import {
     ExperimentParams, ExperimentProfile, Manager, ExperimentStatus,
@@ -24,7 +24,7 @@ import {
     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, IMPORT_DATA
 } from './commands';
-import { createDispatcherInterface, IpcInterface } from './ipcInterface';
+import { createDispatcherInterface, createDispatcherPipeInterface, IpcInterface } from './ipcInterface';
 
 /**
  * NNIManager which implements Manager interface
@@ -71,6 +71,11 @@ class NNIManager implements Manager {
                 this.criticalError(NNIError.FromError(err, 'Job metrics error: '));
             });
         };
+
+        const pipe = getDispatcherPipe();
+        if (pipe !== null) {
+            this.dispatcher = createDispatcherPipeInterface(pipe);
+        }
     }
 
     public updateExperimentProfile(experimentProfile: ExperimentProfile, updateType: ProfileUpdateType): Promise<void> {
@@ -694,7 +699,7 @@ class NNIManager implements Manager {
     }
 
     private async onTrialJobMetrics(metric: TrialJobMetric): Promise<void> {
-        this.log.debug(`NNIManager received trial job metrics: ${metric}`);
+        this.log.debug(`NNIManager received trial job metrics: ${JSON.stringify(metric)}`);
         if (this.trialJobs.has(metric.id)){
             await this.dataStore.storeMetricData(metric.id, metric.data);
             if (this.dispatcher === undefined) {

ts/nni_manager/main.ts

@@ -30,9 +30,9 @@ import { DLTSTrainingService } from './training_service/dlts/dltsTrainingService
 
 function initStartupInfo(
     startExpMode: string, experimentId: string, basePort: number, platform: string,
-    logDirectory: string, experimentLogLevel: string, readonly: boolean): void {
+    logDirectory: string, experimentLogLevel: string, readonly: boolean, dispatcherPipe: string): void {
     const createNew: boolean = (startExpMode === ExperimentStartUpMode.NEW);
-    setExperimentStartupInfo(createNew, experimentId, basePort, platform, logDirectory, experimentLogLevel, readonly);
+    setExperimentStartupInfo(createNew, experimentId, basePort, platform, logDirectory, experimentLogLevel, readonly, dispatcherPipe);
 }
 
 async function initContainer(foreground: boolean, platformMode: string, logFileName?: string): Promise<void> {
@@ -163,7 +163,9 @@ if (!('true' || 'false').includes(readonlyArg.toLowerCase())) {
 }
 const readonly = readonlyArg.toLowerCase() == 'true' ? true : false;
 
-initStartupInfo(startMode, experimentId, port, mode, logDir, logLevel, readonly);
+const dispatcherPipe: string = parseArg(['--dispatcher_pipe']);
+
+initStartupInfo(startMode, experimentId, port, mode, logDir, logLevel, readonly, dispatcherPipe);
 
 mkDirP(getLogDir())
     .then(async () => {

ts/nni_manager/yarn.lock

@@ -279,6 +279,11 @@
   version "7.0.3"
   resolved "https://registry.yarnpkg.com/@types/json-schema/-/json-schema-7.0.3.tgz#bdfd69d61e464dcc81b25159c270d75a73c1a636"
 
+"@types/lockfile@^1.0.0":
+  version "1.0.1"
+  resolved "https://registry.yarnpkg.com/@types/lockfile/-/lockfile-1.0.1.tgz#434a3455e89843312f01976e010c60f1bcbd56f7"
+  integrity sha512-65WZedEm4AnOsBDdsapJJG42MhROu3n4aSSiu87JXF/pSdlubxZxp3S1yz3kTfkJ2KBPud4CpjoHVAptOm9Zmw==
+
 "@types/mime@*":
   version "2.0.0"
   resolved "https://registry.yarnpkg.com/@types/mime/-/mime-2.0.0.tgz#5a7306e367c539b9f6543499de8dd519fac37a8b"