remoteMachineTrainingService.ts 35.7 KB
Newer Older
Deshui Yu's avatar
Deshui Yu committed
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
/**
 * Copyright (c) Microsoft Corporation
 * All rights reserved.
 *
 * MIT License
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
 * documentation files (the "Software"), to deal in the Software without restriction, including without limitation
 * the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and
 * to permit persons to whom the Software is furnished to do so, subject to the following conditions:
 * The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED *AS IS*, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING
 * BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
 * DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 */

'use strict';

22
import * as assert from 'assert';
Deshui Yu's avatar
Deshui Yu committed
23
24
25
26
27
import * as cpp from 'child-process-promise';
import { EventEmitter } from 'events';
import * as fs from 'fs';
import * as os from 'os';
import * as path from 'path';
28
import { Client, ConnectConfig } from 'ssh2';
Deshui Yu's avatar
Deshui Yu committed
29
30
31
import { Deferred } from 'ts-deferred';
import { String } from 'typescript-string-operations';
import * as component from '../../common/component';
SparkSnail's avatar
SparkSnail committed
32
import { NNIError, NNIErrorNames } from '../../common/errors';
33
import { getExperimentId, getInitTrialSequenceId } from '../../common/experimentStartupInfo';
Deshui Yu's avatar
Deshui Yu committed
34
35
36
import { getLogger, Logger } from '../../common/log';
import { ObservableTimer } from '../../common/observableTimer';
import {
SparkSnail's avatar
SparkSnail committed
37
    HostJobApplicationForm, HyperParameters, JobApplicationForm, TrainingService, TrialJobApplicationForm, TrialJobDetail, TrialJobMetric, NNIManagerIpConfig
Deshui Yu's avatar
Deshui Yu committed
38
} from '../../common/trainingService';
39
import { delay, generateParamFileName, getExperimentRootDir, uniqueString, getJobCancelStatus, getRemoteTmpDir,getIPV4Address, getVersion, unixPathJoin } from '../../common/utils';
Deshui Yu's avatar
Deshui Yu committed
40
import { GPUSummary } from '../common/gpuData';
41
42
import { TrialConfig } from '../common/trialConfig';
import { TrialConfigMetadataKey } from '../common/trialConfigMetadataKey';
Deshui Yu's avatar
Deshui Yu committed
43
44
import { GPUScheduler } from './gpuScheduler';
import {
45
    HOST_JOB_SHELL_FORMAT, RemoteCommandResult, RemoteMachineMeta,
SparkSnail's avatar
SparkSnail committed
46
    RemoteMachineScheduleInfo, RemoteMachineScheduleResult, SSHClient, SSHClientManager,
47
    RemoteMachineTrialJobDetail, ScheduleResultType, REMOTEMACHINE_TRIAL_COMMAND_FORMAT
Deshui Yu's avatar
Deshui Yu committed
48
} from './remoteMachineData';
49
import { GPU_INFO_COLLECTOR_FORMAT_LINUX } from '../common/gpuData';
Deshui Yu's avatar
Deshui Yu committed
50
import { SSHClientUtility } from './sshClientUtility';
51
import { validateCodeDir, execRemove, execMkdir, execCopydir } from '../common/util';
SparkSnail's avatar
SparkSnail committed
52
53
import { RemoteMachineJobRestServer } from './remoteMachineJobRestServer';
import { CONTAINER_INSTALL_NNI_SHELL_FORMAT } from '../common/containerJobData';
Deshui Yu's avatar
Deshui Yu committed
54
55
56
57

/**
 * Training Service implementation for Remote Machine (Linux)
 */
SparkSnail's avatar
SparkSnail committed
58
@component.Singleton
Deshui Yu's avatar
Deshui Yu committed
59
class RemoteMachineTrainingService implements TrainingService {
SparkSnail's avatar
SparkSnail committed
60
61
    private machineSSHClientMap: Map<RemoteMachineMeta, SSHClientManager>; //machine ssh client map
    private trialSSHClientMap: Map<string, Client>; //trial ssh client map
Deshui Yu's avatar
Deshui Yu committed
62
    private trialJobsMap: Map<string, RemoteMachineTrialJobDetail>;
SparkSnail's avatar
SparkSnail committed
63
    private readonly MAX_TRIAL_NUMBER_PER_SSHCONNECTION: number = 5 // every ssh client has a max trial concurrency number
Deshui Yu's avatar
Deshui Yu committed
64
65
    private expRootDir: string;
    private remoteExpRootDir: string;
66
    private trialConfig: TrialConfig | undefined;
Deshui Yu's avatar
Deshui Yu committed
67
68
69
70
71
72
    private gpuScheduler: GPUScheduler;
    private jobQueue: string[];
    private timer: ObservableTimer;
    private stopping: boolean = false;
    private metricsEmitter: EventEmitter;
    private log: Logger;
73
    private isMultiPhase: boolean = false;
74
    private trialSequenceId: number;
SparkSnail's avatar
SparkSnail committed
75
    private remoteRestServerPort?: number;
76
    private readonly remoteOS: string;
SparkSnail's avatar
SparkSnail committed
77
    private nniManagerIpConfig?: NNIManagerIpConfig;
78
    private versionCheck: boolean = true;
SparkSnail's avatar
SparkSnail committed
79
    private logCollection: string;
Deshui Yu's avatar
Deshui Yu committed
80
81

    constructor(@component.Inject timer: ObservableTimer) {
82
        this.remoteOS = 'linux';
Deshui Yu's avatar
Deshui Yu committed
83
84
        this.metricsEmitter = new EventEmitter();
        this.trialJobsMap = new Map<string, RemoteMachineTrialJobDetail>();
SparkSnail's avatar
SparkSnail committed
85
86
        this.trialSSHClientMap = new Map<string, Client>();
        this.machineSSHClientMap = new Map<RemoteMachineMeta, SSHClientManager>();
Deshui Yu's avatar
Deshui Yu committed
87
88
89
        this.gpuScheduler = new GPUScheduler(this.machineSSHClientMap);
        this.jobQueue = [];
        this.expRootDir = getExperimentRootDir();
90
        this.remoteExpRootDir = this.getRemoteExperimentRootDir();
Deshui Yu's avatar
Deshui Yu committed
91
92
        this.timer = timer;
        this.log = getLogger();
93
        this.trialSequenceId = -1;
SparkSnail's avatar
SparkSnail committed
94
        this.logCollection = 'none';
chicm-ms's avatar
chicm-ms committed
95
        this.log.info('Construct remote machine training service.');
Deshui Yu's avatar
Deshui Yu committed
96
97
98
99
100
101
    }

    /**
     * Loop to launch trial jobs and collect trial metrics
     */
    public async run(): Promise<void> {
SparkSnail's avatar
SparkSnail committed
102
103
        const restServer: RemoteMachineJobRestServer = component.get(RemoteMachineJobRestServer);
        await restServer.start();
104
        restServer.setEnableVersionCheck = this.versionCheck;
chicm-ms's avatar
chicm-ms committed
105
        this.log.info('Run remote machine training service.');
Deshui Yu's avatar
Deshui Yu committed
106
107
        while (!this.stopping) {
            while (this.jobQueue.length > 0) {
SparkSnail's avatar
SparkSnail committed
108
                this.updateGpuReservation();
Deshui Yu's avatar
Deshui Yu committed
109
110
111
112
113
114
                const trialJobId: string = this.jobQueue[0];
                const prepareResult : boolean = await this.prepareTrialJob(trialJobId);
                if (prepareResult) {
                    // Remove trial job with trialJobId from job queue
                    this.jobQueue.shift();
                } else {
115
                    // Break the while loop since no GPU resource is available right now,
Deshui Yu's avatar
Deshui Yu committed
116
117
118
                    // Wait to schedule job in next time iteration
                    break;
                }
119
            }
120
121
122
123
            if(restServer.getErrorMessage) {
                throw new Error(restServer.getErrorMessage);
                this.stopping = true;
            }
Deshui Yu's avatar
Deshui Yu committed
124
125
            await delay(3000);
        }
chicm-ms's avatar
chicm-ms committed
126
        this.log.info('Remote machine training service exit.');
Deshui Yu's avatar
Deshui Yu committed
127
    }
128

SparkSnail's avatar
SparkSnail committed
129
130
    /**
     * give trial a ssh connection
131
     * @param trial
SparkSnail's avatar
SparkSnail committed
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
     */
    public async allocateSSHClientForTrial(trial: RemoteMachineTrialJobDetail): Promise<void> {
        const deferred: Deferred<void> = new Deferred<void>();
        if(!trial.rmMeta) {
            throw new Error(`rmMeta not set in trial ${trial.id}`);
        }
        let sshClientManager: SSHClientManager | undefined = this.machineSSHClientMap.get(trial.rmMeta);
        if(!sshClientManager) {
            throw new Error(`remoteSSHClient not initialized`);
        }
        let sshClient: Client = await sshClientManager.getAvailableSSHClient();
        this.trialSSHClientMap.set(trial.id, sshClient);
        deferred.resolve();
        return deferred.promise;
    }
147

SparkSnail's avatar
SparkSnail committed
148
149
    /**
     * If a trial is finished, release the connection resource
150
     * @param trial
SparkSnail's avatar
SparkSnail committed
151
152
153
154
155
156
157
158
159
160
161
     */
    public releaseTrialSSHClient(trial: RemoteMachineTrialJobDetail): void {
        if(!trial.rmMeta) {
            throw new Error(`rmMeta not set in trial ${trial.id}`);
        }
        let sshClientManager: SSHClientManager | undefined = this.machineSSHClientMap.get(trial.rmMeta);
        if(!sshClientManager) {
            throw new Error(`sshClientManager not initialized`);
        }
        sshClientManager.releaseConnection(this.trialSSHClientMap.get(trial.id));
    }
Deshui Yu's avatar
Deshui Yu committed
162
163
164
165

    /**
     * List submitted trial jobs
     */
166
    public async listTrialJobs(): Promise<TrialJobDetail[]> {
Deshui Yu's avatar
Deshui Yu committed
167
168
169
        const jobs: TrialJobDetail[] = [];
        const deferred: Deferred<TrialJobDetail[]> = new Deferred<TrialJobDetail[]>();

170
        for (const [key, value] of this.trialJobsMap) {
Deshui Yu's avatar
Deshui Yu committed
171
172
173
            if (value.form.jobType === 'TRIAL') {
                jobs.push(await this.getTrialJob(key));
            }
174
        };
Deshui Yu's avatar
Deshui Yu committed
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
        deferred.resolve(jobs);

        return deferred.promise;
    }

    /**
     * Get trial job detail information
     * @param trialJobId ID of trial job
     */
    public async getTrialJob(trialJobId: string): Promise<TrialJobDetail> {
        const trialJob: RemoteMachineTrialJobDetail | undefined = this.trialJobsMap.get(trialJobId);
        if (!trialJob) {
            throw new NNIError(NNIErrorNames.NOT_FOUND, `trial job id ${trialJobId} not found`);
        }
        //TO DO: add another job status, and design new job status change logic
        if (trialJob.status === 'RUNNING' || trialJob.status === 'UNKNOWN') {
            // Get ssh client where the job is running
            if (trialJob.rmMeta === undefined) {
                throw new Error(`rmMeta not set for submitted job ${trialJobId}`);
            }
SparkSnail's avatar
SparkSnail committed
195
            const sshClient: Client | undefined  = this.trialSSHClientMap.get(trialJob.id);
Deshui Yu's avatar
Deshui Yu committed
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
            if (!sshClient) {
                throw new Error(`Invalid job id: ${trialJobId}, cannot find ssh client`);
            }

            return this.updateTrialJobStatus(trialJob, sshClient);
        } else {
            return trialJob;
        }
    }

    /**
     * Add job metrics listener
     * @param listener callback listener
     */
    public addTrialJobMetricListener(listener: (metric: TrialJobMetric) => void): void {
        this.metricsEmitter.on('metric', listener);
    }

    /**
     * Remove job metrics listener
     * @param listener callback listener
     */
    public removeTrialJobMetricListener(listener: (metric: TrialJobMetric) => void): void {
        this.metricsEmitter.off('metric', listener);
    }

    /**
     * Submit trial job
     * @param form trial job description form
     */
SparkSnail's avatar
SparkSnail committed
226
    public async submitTrialJob(form: JobApplicationForm): Promise<TrialJobDetail> {
Deshui Yu's avatar
Deshui Yu committed
227
228
229
230
231
232
233
234
235
        if (!this.trialConfig) {
            throw new Error('trial config is not initialized');
        }

        if (form.jobType === 'HOST') {
            return this.runHostJob(<HostJobApplicationForm>form);
        } else if (form.jobType === 'TRIAL') {
            // Generate trial job id(random)
            const trialJobId: string = uniqueString(5);
236
            const trialWorkingFolder: string = unixPathJoin(this.remoteExpRootDir, 'trials', trialJobId);
Deshui Yu's avatar
Deshui Yu committed
237
238
239
240

            const trialJobDetail: RemoteMachineTrialJobDetail = new RemoteMachineTrialJobDetail(
                trialJobId,
                'WAITING',
241
                Date.now(),
Deshui Yu's avatar
Deshui Yu committed
242
                trialWorkingFolder,
243
244
245
                form,
                this.generateSequenceId()
            );
Deshui Yu's avatar
Deshui Yu committed
246
247
            this.jobQueue.push(trialJobId);
            this.trialJobsMap.set(trialJobId, trialJobDetail);
248

Deshui Yu's avatar
Deshui Yu committed
249
250
251
252
253
254
            return Promise.resolve(trialJobDetail);
        } else {
            return Promise.reject(new Error(`Job form not supported: ${JSON.stringify(form)}, jobType should be HOST or TRIAL.`));
        }
    }

255
256
257
258
259
    /**
     * Update trial job for multi-phase
     * @param trialJobId trial job id
     * @param form job application form
     */
chicm-ms's avatar
chicm-ms committed
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
    public async updateTrialJob(trialJobId: string, form: JobApplicationForm): Promise<TrialJobDetail> {
        const trialJobDetail: undefined | TrialJobDetail = this.trialJobsMap.get(trialJobId);
        if (trialJobDetail === undefined) {
            throw new Error(`updateTrialJob failed: ${trialJobId} not found`);
        }
        if (form.jobType === 'TRIAL') {
            const rmMeta: RemoteMachineMeta | undefined = (<RemoteMachineTrialJobDetail>trialJobDetail).rmMeta;
            if (rmMeta !== undefined) {
                await this.writeParameterFile(trialJobId, (<TrialJobApplicationForm>form).hyperParameters, rmMeta);
            } else {
                throw new Error(`updateTrialJob failed: ${trialJobId} rmMeta not found`);
            }
        } else {
            throw new Error(`updateTrialJob failed: jobType ${form.jobType} not supported.`);
        }

        return trialJobDetail;
277
    }
278

SparkSnail's avatar
SparkSnail committed
279
280
281
282
    /**
     * remove gpu reversion when job is not running
     */
    private updateGpuReservation() {
283
        for (const [key, value] of this.trialJobsMap) {
SparkSnail's avatar
SparkSnail committed
284
            if(!['WAITING', 'RUNNING'].includes(value.status)) {
285
                this.gpuScheduler.removeGpuReservation(key, this.trialJobsMap);
SparkSnail's avatar
SparkSnail committed
286
287
288
            }
        };
    }
289
290
291
292
293

    /**
     * Is multiphase job supported in current training service
     */
    public get isMultiPhaseJobSupported(): boolean {
294
        return true;
295
296
    }

Deshui Yu's avatar
Deshui Yu committed
297
298
299
300
    /**
     * Cancel trial job
     * @param trialJobId ID of trial job
     */
QuanluZhang's avatar
QuanluZhang committed
301
    public async cancelTrialJob(trialJobId: string, isEarlyStopped: boolean = false): Promise<void> {
Deshui Yu's avatar
Deshui Yu committed
302
303
304
305
306
307
308
309
310
        const deferred: Deferred<void> = new Deferred<void>();
        const trialJob: RemoteMachineTrialJobDetail | undefined = this.trialJobsMap.get(trialJobId);
        if (!trialJob) {
            deferred.reject();
            throw new Error(`trial job id ${trialJobId} not found`);
        }

        // Remove the job with trialJobId from job queue
        const index : number = this.jobQueue.indexOf(trialJobId);
311
        if (index >= 0) {
Deshui Yu's avatar
Deshui Yu committed
312
313
314
315
316
317
            this.jobQueue.splice(index, 1);
        }

        // Get ssh client where the job is running
        if (trialJob.rmMeta !== undefined) {
            // If the trial job is already scheduled, check its status and kill the trial process in remote machine
SparkSnail's avatar
SparkSnail committed
318
            const sshClient: Client | undefined = this.trialSSHClientMap.get(trialJob.id);
Deshui Yu's avatar
Deshui Yu committed
319
320
321
322
323
324
325
            if (!sshClient) {
                deferred.reject();
                throw new Error(`Invalid job id ${trialJobId}, cannot find ssh client`);
            }

            const jobpidPath: string = this.getJobPidPath(trialJob.id);
            try {
326
327
                // Mark the toEarlyStop tag here
                trialJob.isEarlyStopped = isEarlyStopped;
Deshui Yu's avatar
Deshui Yu committed
328
                await SSHClientUtility.remoteExeCommand(`pkill -P \`cat ${jobpidPath}\``, sshClient);
SparkSnail's avatar
SparkSnail committed
329
                this.releaseTrialSSHClient(trialJob);
Deshui Yu's avatar
Deshui Yu committed
330
331
332
333
334
335
            } catch (error) {
                // Not handle the error since pkill failed will not impact trial job's current status
                this.log.error(`remoteTrainingService.cancelTrialJob: ${error.message}`);
            }
        } else {
            // Job is not scheduled yet, set status to 'USER_CANCELLED' directly
QuanluZhang's avatar
QuanluZhang committed
336
337
            assert(isEarlyStopped === false, 'isEarlyStopped is not supposed to be true here.');
            trialJob.status = getJobCancelStatus(isEarlyStopped);
Deshui Yu's avatar
Deshui Yu committed
338
339
340
341
342
343
344
345
346
347
348
349
        }
    }

    /**
     * Set culster metadata
     * @param key metadata key
     * //1. MACHINE_LIST -- create ssh client connect of machine list
     * //2. TRIAL_CONFIG -- trial configuration
     * @param value metadata value
     */
    public async setClusterMetadata(key: string, value: string): Promise<void> {
        switch (key) {
SparkSnail's avatar
SparkSnail committed
350
351
352
            case TrialConfigMetadataKey.NNI_MANAGER_IP:
                this.nniManagerIpConfig = <NNIManagerIpConfig>JSON.parse(value);
                break;
353
            case TrialConfigMetadataKey.MACHINE_LIST:
Deshui Yu's avatar
Deshui Yu committed
354
                await this.setupConnections(value);
SparkSnail's avatar
SparkSnail committed
355
                //remove local temp files
356
                await execRemove(this.getLocalGpuMetricCollectorDir());
Deshui Yu's avatar
Deshui Yu committed
357
                break;
358
359
            case TrialConfigMetadataKey.TRIAL_CONFIG:
                const remoteMachineTrailConfig: TrialConfig = <TrialConfig>JSON.parse(value);
Deshui Yu's avatar
Deshui Yu committed
360
361
362
363
364
365
366
367
                // Parse trial config failed, throw Error
                if (!remoteMachineTrailConfig) {
                    throw new Error('trial config parsed failed');
                }
                // codeDir is not a valid directory, throw Error
                if (!fs.lstatSync(remoteMachineTrailConfig.codeDir).isDirectory()) {
                    throw new Error(`codeDir ${remoteMachineTrailConfig.codeDir} is not a directory`);
                }
368
369
370
371
372
373

                // Validate to make sure codeDir doesn't have too many files
                try {
                    await validateCodeDir(remoteMachineTrailConfig.codeDir);
                } catch(error) {
                    this.log.error(error);
374
                    return Promise.reject(new Error(error));
375
376
                }

Deshui Yu's avatar
Deshui Yu committed
377
378
                this.trialConfig = remoteMachineTrailConfig;
                break;
379
380
381
            case TrialConfigMetadataKey.MULTI_PHASE:
                this.isMultiPhase = (value === 'true' || value === 'True');
                break;
382
383
384
            case TrialConfigMetadataKey.VERSION_CHECK:
                this.versionCheck = (value === 'true' || value === 'True');
                break;
SparkSnail's avatar
SparkSnail committed
385
386
387
            case TrialConfigMetadataKey.LOG_COLLECTION:
                this.logCollection = value;
                break;
Deshui Yu's avatar
Deshui Yu committed
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
            default:
                //Reject for unknown keys
                throw new Error(`Uknown key: ${key}`);
        }
    }

    /**
     * Get culster metadata
     * @param key metadata key
     */
    public getClusterMetadata(key: string): Promise<string> {
        const deferred: Deferred<string> = new Deferred<string>();

        return deferred.promise;
    }
403

SparkSnail's avatar
SparkSnail committed
404
    /**
405
     * cleanup() has a time out of 10s to clean remote connections
SparkSnail's avatar
SparkSnail committed
406
407
     */
    public async cleanUp(): Promise<void> {
chicm-ms's avatar
chicm-ms committed
408
        this.log.info('Stopping remote machine training service...');
Deshui Yu's avatar
Deshui Yu committed
409
        this.stopping = true;
SparkSnail's avatar
SparkSnail committed
410
411
        await Promise.race([delay(10000), this.cleanupConnections()]);
    }
412

SparkSnail's avatar
SparkSnail committed
413
414
415
416
417
    /**
     * stop gpu_metric_collector process in remote machine and remove unused scripts
     */
    private async cleanupConnections(): Promise<void> {
        try{
SparkSnail's avatar
SparkSnail committed
418
            for (const [rmMeta, sshClientManager] of this.machineSSHClientMap.entries()) {
419
                let jobpidPath: string = unixPathJoin(this.getRemoteScriptsPath(rmMeta.username), 'pid');
SparkSnail's avatar
SparkSnail committed
420
421
422
423
424
425
                let client: Client | undefined = sshClientManager.getFirstSSHClient();
                if(client) {
                    await SSHClientUtility.remoteExeCommand(`pkill -P \`cat ${jobpidPath}\``, client);
                    await SSHClientUtility.remoteExeCommand(`rm -rf ${this.getRemoteScriptsPath(rmMeta.username)}`, client);
                }
                sshClientManager.closeAllSSHClient();
SparkSnail's avatar
SparkSnail committed
426
427
428
429
430
            }
        }catch (error) {
            //ignore error, this function is called to cleanup remote connections when experiment is stopping
            this.log.error(`Cleanup connection exception, error is ${error.message}`);
        }
Deshui Yu's avatar
Deshui Yu committed
431
432

        return Promise.resolve();
433
434
    }

SparkSnail's avatar
SparkSnail committed
435
436
437
438
439
    /**
     * Generate gpu metric collector directory to store temp gpu metric collector script files
     */
    private getLocalGpuMetricCollectorDir(): string {
        let userName: string = path.basename(os.homedir()); //get current user name of os
440
        return path.join(os.tmpdir(), userName, 'nni', 'scripts');
SparkSnail's avatar
SparkSnail committed
441
442
443
    }

    /**
444
445
     * Generate gpu metric collector shell script in local machine,
     * used to run in remote machine, and will be deleted after uploaded from local.
SparkSnail's avatar
SparkSnail committed
446
447
448
     */
    private async generateGpuMetricsCollectorScript(userName: string): Promise<void> {
        let gpuMetricCollectorScriptFolder : string = this.getLocalGpuMetricCollectorDir();
449
        await execMkdir(path.join(gpuMetricCollectorScriptFolder, userName));
SparkSnail's avatar
SparkSnail committed
450
451
452
453
        //generate gpu_metrics_collector.sh
        let gpuMetricsCollectorScriptPath: string = path.join(gpuMetricCollectorScriptFolder, userName, 'gpu_metrics_collector.sh');
        const remoteGPUScriptsDir: string = this.getRemoteScriptsPath(userName); // This directory is used to store gpu_metrics and pid created by script
        const gpuMetricsCollectorScriptContent: string = String.Format(
454
455
456
            GPU_INFO_COLLECTOR_FORMAT_LINUX,
            remoteGPUScriptsDir,
            unixPathJoin(remoteGPUScriptsDir, 'pid'),
SparkSnail's avatar
SparkSnail committed
457
458
        );
        await fs.promises.writeFile(gpuMetricsCollectorScriptPath, gpuMetricsCollectorScriptContent, { encoding: 'utf8' });
Deshui Yu's avatar
Deshui Yu committed
459
460
461
    }

    private async setupConnections(machineList: string): Promise<void> {
chicm-ms's avatar
chicm-ms committed
462
        this.log.debug(`Connecting to remote machines: ${machineList}`);
Deshui Yu's avatar
Deshui Yu committed
463
464
465
466
        const deferred: Deferred<void> = new Deferred<void>();
        //TO DO: verify if value's format is wrong, and json parse failed, how to handle error
        const rmMetaList: RemoteMachineMeta[] = <RemoteMachineMeta[]>JSON.parse(machineList);
        let connectedRMNum: number = 0;
SparkSnail's avatar
SparkSnail committed
467

SparkSnail's avatar
SparkSnail committed
468
        rmMetaList.forEach(async (rmMeta: RemoteMachineMeta) => {
469
            rmMeta.occupiedGpuIndexMap = new Map<number, number>();
SparkSnail's avatar
SparkSnail committed
470
471
472
473
474
475
            let sshClientManager: SSHClientManager = new SSHClientManager([], this.MAX_TRIAL_NUMBER_PER_SSHCONNECTION, rmMeta);
            let sshClient: Client = await sshClientManager.getAvailableSSHClient();
            this.machineSSHClientMap.set(rmMeta, sshClientManager);
            await this.initRemoteMachineOnConnected(rmMeta, sshClient);
            if (++connectedRMNum === rmMetaList.length) {
                deferred.resolve();
476
            }
Deshui Yu's avatar
Deshui Yu committed
477
478
479
480
481
482
        });
        return deferred.promise;
    }

    private async initRemoteMachineOnConnected(rmMeta: RemoteMachineMeta, conn: Client): Promise<void> {
        // Create root working directory after ssh connection is ready
SparkSnail's avatar
SparkSnail committed
483
        await this.generateGpuMetricsCollectorScript(rmMeta.username); //generate gpu script in local machine first, will copy to remote machine later
484
        const nniRootDir: string = unixPathJoin(getRemoteTmpDir(this.remoteOS), 'nni');
Deshui Yu's avatar
Deshui Yu committed
485
        await SSHClientUtility.remoteExeCommand(`mkdir -p ${this.remoteExpRootDir}`, conn);
486

Deshui Yu's avatar
Deshui Yu committed
487
        // Copy NNI scripts to remote expeirment working directory
SparkSnail's avatar
SparkSnail committed
488
489
490
        const localGpuScriptCollectorDir: string = this.getLocalGpuMetricCollectorDir();
        const remoteGpuScriptCollectorDir: string = this.getRemoteScriptsPath(rmMeta.username); //the directory to store temp scripts in remote machine
        await SSHClientUtility.remoteExeCommand(`mkdir -p ${remoteGpuScriptCollectorDir}`, conn);
Deshui Yu's avatar
Deshui Yu committed
491
        await SSHClientUtility.remoteExeCommand(`chmod 777 ${nniRootDir} ${nniRootDir}/* ${nniRootDir}/scripts/*`, conn);
SparkSnail's avatar
SparkSnail committed
492
        //copy gpu_metrics_collector.sh to remote
493
        await SSHClientUtility.copyFileToRemote(path.join(localGpuScriptCollectorDir, rmMeta.username, 'gpu_metrics_collector.sh'), unixPathJoin(remoteGpuScriptCollectorDir, 'gpu_metrics_collector.sh'), conn);
494

Deshui Yu's avatar
Deshui Yu committed
495
        //Begin to execute gpu_metrics_collection scripts
496
        SSHClientUtility.remoteExeCommand(`bash ${unixPathJoin(remoteGpuScriptCollectorDir, 'gpu_metrics_collector.sh')}`, conn);
497

Deshui Yu's avatar
Deshui Yu committed
498
499
500
        this.timer.subscribe(
            async (tick: number) => {
                const cmdresult: RemoteCommandResult = await SSHClientUtility.remoteExeCommand(
501
                    `tail -n 1 ${unixPathJoin(remoteGpuScriptCollectorDir, 'gpu_metrics')}`, conn);
Deshui Yu's avatar
Deshui Yu committed
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
                if (cmdresult && cmdresult.stdout) {
                    rmMeta.gpuSummary = <GPUSummary>JSON.parse(cmdresult.stdout);
                }
            }
        );
    }

    private async prepareTrialJob(trialJobId: string): Promise<boolean> {
        const deferred : Deferred<boolean> = new Deferred<boolean>();

        if (!this.trialConfig) {
            throw new Error('trial config is not initialized');
        }
        const trialJobDetail: RemoteMachineTrialJobDetail | undefined = this.trialJobsMap.get(trialJobId);
        if (trialJobDetail === undefined) {
            throw new NNIError(NNIErrorNames.INVALID_JOB_DETAIL, `Invalid job detail information for trial job ${trialJobId}`);
        }
519
520
521
522
523
        // If job is not WATIING, Don't prepare and resolve true immediately
        if (trialJobDetail.status !== 'WAITING') {
            deferred.resolve(true);
            return deferred.promise;
        }
Deshui Yu's avatar
Deshui Yu committed
524
        // get an ssh client from scheduler
525
        const rmScheduleResult: RemoteMachineScheduleResult = this.gpuScheduler.scheduleMachine(this.trialConfig.gpuNum, trialJobDetail);
Deshui Yu's avatar
Deshui Yu committed
526
527
528
529
530
        if (rmScheduleResult.resultType === ScheduleResultType.REQUIRE_EXCEED_TOTAL) {
            const errorMessage : string = `Required GPU number ${this.trialConfig.gpuNum} is too large, no machine can meet`;
            this.log.error(errorMessage);
            deferred.reject();
            throw new NNIError(NNIErrorNames.RESOURCE_NOT_AVAILABLE, errorMessage);
531
        } else if (rmScheduleResult.resultType === ScheduleResultType.SUCCEED
Deshui Yu's avatar
Deshui Yu committed
532
533
            && rmScheduleResult.scheduleInfo !== undefined) {
            const rmScheduleInfo : RemoteMachineScheduleInfo = rmScheduleResult.scheduleInfo;
534
            const trialWorkingFolder: string = unixPathJoin(this.remoteExpRootDir, 'trials', trialJobId);
SparkSnail's avatar
SparkSnail committed
535
536
537
538

            trialJobDetail.rmMeta = rmScheduleInfo.rmMeta;

            await this.allocateSSHClientForTrial(trialJobDetail);
Deshui Yu's avatar
Deshui Yu committed
539
540
541
542
543
            await this.launchTrialOnScheduledMachine(
                trialJobId, trialWorkingFolder, <TrialJobApplicationForm>trialJobDetail.form, rmScheduleInfo);

            trialJobDetail.status = 'RUNNING';
            trialJobDetail.url = `file://${rmScheduleInfo.rmMeta.ip}:${trialWorkingFolder}`;
544
            trialJobDetail.startTime = Date.now();
Deshui Yu's avatar
Deshui Yu committed
545

546
            this.trialJobsMap.set(trialJobId, trialJobDetail);
Deshui Yu's avatar
Deshui Yu committed
547
            deferred.resolve(true);
548
        } else if (rmScheduleResult.resultType === ScheduleResultType.TMP_NO_AVAILABLE_GPU) {
Deshui Yu's avatar
Deshui Yu committed
549
550
551
            this.log.info(`Right now no available GPU can be allocated for trial ${trialJobId}, will try to schedule later`);
            deferred.resolve(false);
        } else {
552
            deferred.reject(`Invalid schedule resutl type: ${rmScheduleResult.resultType}`);
Deshui Yu's avatar
Deshui Yu committed
553
554
555
556
557
558
559
560
561
562
563
        }

        return deferred.promise;
    }

    private async launchTrialOnScheduledMachine(trialJobId: string, trialWorkingFolder: string, form: TrialJobApplicationForm,
                                                rmScheduleInfo: RemoteMachineScheduleInfo): Promise<void> {
        if (!this.trialConfig) {
            throw new Error('trial config is not initialized');
        }
        const cuda_visible_device: string = rmScheduleInfo.cuda_visible_device;
SparkSnail's avatar
SparkSnail committed
564
        const sshClient: Client | undefined = this.trialSSHClientMap.get(trialJobId);
565
566
567
568
569
570
        if (sshClient === undefined) {
            assert(false, 'sshClient is undefined.');

            // for lint
            return;
        }
571
572
573
574
575
        const trialJobDetail: RemoteMachineTrialJobDetail | undefined = this.trialJobsMap.get(trialJobId);
        if (trialJobDetail === undefined) {
            throw new Error(`Can not get trial job detail for job: ${trialJobId}`);
        }

Deshui Yu's avatar
Deshui Yu committed
576
577
578
        const trialLocalTempFolder: string = path.join(this.expRootDir, 'trials-local', trialJobId);

        await SSHClientUtility.remoteExeCommand(`mkdir -p ${trialWorkingFolder}`, sshClient);
579
        await SSHClientUtility.remoteExeCommand(`mkdir -p ${unixPathJoin(trialWorkingFolder, '.nni')}`, sshClient);
Deshui Yu's avatar
Deshui Yu committed
580
581
582

        // RemoteMachineRunShellFormat is the run shell format string,
        // See definition in remoteMachineData.ts
SparkSnail's avatar
SparkSnail committed
583
584
585
586
587
588
589
590
591

        let command: string;
        // Set CUDA_VISIBLE_DEVICES environment variable based on cuda_visible_device
        // If no valid cuda_visible_device is defined, set CUDA_VISIBLE_DEVICES to empty string to hide GPU device
        if(typeof cuda_visible_device === 'string' && cuda_visible_device.length > 0) {
            command = `CUDA_VISIBLE_DEVICES=${cuda_visible_device} ${this.trialConfig.command}`;
        } else {
            command = `CUDA_VISIBLE_DEVICES=" " ${this.trialConfig.command}`;
        }
592

SparkSnail's avatar
SparkSnail committed
593
594
595
596
597
        const nniManagerIp = this.nniManagerIpConfig?this.nniManagerIpConfig.nniManagerIp:getIPV4Address();
        if(!this.remoteRestServerPort) {
            const restServer: RemoteMachineJobRestServer = component.get(RemoteMachineJobRestServer);
            this.remoteRestServerPort = restServer.clusterRestServerPort;
        }
598
        const version = this.versionCheck? await getVersion(): '';
SparkSnail's avatar
SparkSnail committed
599
600
601
        const runScriptTrialContent: string = String.Format(
            REMOTEMACHINE_TRIAL_COMMAND_FORMAT,
            trialWorkingFolder,
Deshui Yu's avatar
Deshui Yu committed
602
603
            trialWorkingFolder,
            trialJobId,
SparkSnail's avatar
SparkSnail committed
604
605
            getExperimentId(),
            trialJobDetail.sequenceId.toString(),
606
            this.isMultiPhase,
607
            unixPathJoin(trialWorkingFolder, '.nni', 'jobpid'),
SparkSnail's avatar
SparkSnail committed
608
609
610
            command,
            nniManagerIp,
            this.remoteRestServerPort,
611
            version,
SparkSnail's avatar
SparkSnail committed
612
            this.logCollection,
613
            unixPathJoin(trialWorkingFolder, '.nni', 'code')
SparkSnail's avatar
SparkSnail committed
614
        )
Deshui Yu's avatar
Deshui Yu committed
615
616

        //create tmp trial working folder locally.
617
        await execMkdir(path.join(trialLocalTempFolder, '.nni'));
Deshui Yu's avatar
Deshui Yu committed
618

SparkSnail's avatar
SparkSnail committed
619
        //create tmp trial working folder locally.
620
        await execCopydir(path.join(this.trialConfig.codeDir, '*'), trialLocalTempFolder);
SparkSnail's avatar
SparkSnail committed
621
622
623
        const installScriptContent : string = CONTAINER_INSTALL_NNI_SHELL_FORMAT;
        // Write NNI installation file to local tmp files
        await fs.promises.writeFile(path.join(trialLocalTempFolder, 'install_nni.sh'), installScriptContent, { encoding: 'utf8' });
624
        // Write file content ( run.sh and parameter.cfg ) to local tmp files
SparkSnail's avatar
SparkSnail committed
625
        await fs.promises.writeFile(path.join(trialLocalTempFolder, 'run.sh'), runScriptTrialContent, { encoding: 'utf8' });
chicm-ms's avatar
chicm-ms committed
626
        await this.writeParameterFile(trialJobId, form.hyperParameters, rmScheduleInfo.rmMeta);
Deshui Yu's avatar
Deshui Yu committed
627
        // Copy files in codeDir to remote working directory
SparkSnail's avatar
SparkSnail committed
628
        await SSHClientUtility.copyDirectoryToRemote(trialLocalTempFolder, trialWorkingFolder, sshClient, this.remoteOS);
Deshui Yu's avatar
Deshui Yu committed
629
        // Execute command in remote machine
630
        SSHClientUtility.remoteExeCommand(`bash ${unixPathJoin(trialWorkingFolder, 'run.sh')}`, sshClient);
Deshui Yu's avatar
Deshui Yu committed
631
632
633
634
    }

    private async runHostJob(form: HostJobApplicationForm): Promise<TrialJobDetail> {
        const rmMeta: RemoteMachineMeta = this.getRmMetaByHost(form.host);
SparkSnail's avatar
SparkSnail committed
635
636
        const sshClientManager: SSHClientManager | undefined = this.machineSSHClientMap.get(rmMeta);
        if (sshClientManager === undefined) {
Deshui Yu's avatar
Deshui Yu committed
637
638
            throw new Error('sshClient not found.');
        }
SparkSnail's avatar
SparkSnail committed
639
        let sshClient: Client = sshClientManager.getFirstSSHClient();
Deshui Yu's avatar
Deshui Yu committed
640
641
642
643
644
645
        const jobId: string = uniqueString(5);
        const localDir: string = path.join(this.expRootDir, 'hostjobs-local', jobId);
        const remoteDir: string = this.getHostJobRemoteDir(jobId);
        await cpp.exec(`mkdir -p ${localDir}`);
        await SSHClientUtility.remoteExeCommand(`mkdir -p ${remoteDir}`, sshClient);
        const runScriptContent: string = String.Format(
646
            HOST_JOB_SHELL_FORMAT, remoteDir, path.join(remoteDir, 'jobpid'), form.cmd, path.join(remoteDir, 'code')
Deshui Yu's avatar
Deshui Yu committed
647
648
649
        );
        await fs.promises.writeFile(path.join(localDir, 'run.sh'), runScriptContent, { encoding: 'utf8' });
        await SSHClientUtility.copyFileToRemote(
650
651
            path.join(localDir, 'run.sh'), unixPathJoin(remoteDir, 'run.sh'), sshClient);
        SSHClientUtility.remoteExeCommand(`bash ${unixPathJoin(remoteDir, 'run.sh')}`, sshClient);
Deshui Yu's avatar
Deshui Yu committed
652

653
654
655
        const jobDetail: RemoteMachineTrialJobDetail =  new RemoteMachineTrialJobDetail(
            jobId, 'RUNNING', Date.now(), remoteDir, form, this.generateSequenceId()
        );
Deshui Yu's avatar
Deshui Yu committed
656
        jobDetail.rmMeta = rmMeta;
657
        jobDetail.startTime = Date.now();
Deshui Yu's avatar
Deshui Yu committed
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
        this.trialJobsMap.set(jobId, jobDetail);
        this.log.debug(`runHostJob: return: ${JSON.stringify(jobDetail)} `);

        return jobDetail;
    }

    private getRmMetaByHost(host: string): RemoteMachineMeta {
        for (const [rmMeta, client] of this.machineSSHClientMap.entries()) {
            if (rmMeta.ip === host) {
                return rmMeta;
            }
        }
        throw new Error(`Host not found: ${host}`);
    }

    private async updateTrialJobStatus(trialJob: RemoteMachineTrialJobDetail, sshClient: Client): Promise<TrialJobDetail> {
        const deferred: Deferred<TrialJobDetail> = new Deferred<TrialJobDetail>();
        const jobpidPath: string = this.getJobPidPath(trialJob.id);
676
        const trialReturnCodeFilePath: string = unixPathJoin(this.remoteExpRootDir, 'trials', trialJob.id, '.nni', 'code');
Deshui Yu's avatar
Deshui Yu committed
677
678
679
680
681
682
683
684
685
686
687
688
689
        try {
            const killResult: number = (await SSHClientUtility.remoteExeCommand(`kill -0 \`cat ${jobpidPath}\``, sshClient)).exitCode;
            // if the process of jobpid is not alive any more
            if (killResult !== 0) {
                const trailReturnCode: string = await SSHClientUtility.getRemoteFileContent(trialReturnCodeFilePath, sshClient);
                this.log.debug(`trailjob ${trialJob.id} return code: ${trailReturnCode}`);
                const match: RegExpMatchArray | null = trailReturnCode.trim().match(/^(\d+)\s+(\d+)$/);
                if (match) {
                    const { 1: code, 2: timestamp } = match;
                    // Update trial job's status based on result code
                    if (parseInt(code, 10) === 0) {
                        trialJob.status = 'SUCCEEDED';
                    } else {
690
691
692
693
694
695
                        // isEarlyStopped is never set, mean it's not cancelled by NNI, so if the process's exit code >0, mark it as FAILED
                        if (trialJob.isEarlyStopped === undefined) {
                            trialJob.status = 'FAILED';
                        } else {
                            trialJob.status = getJobCancelStatus(trialJob.isEarlyStopped);
                        }
Deshui Yu's avatar
Deshui Yu committed
696
                    }
697
                    trialJob.endTime = parseInt(timestamp, 10);
SparkSnail's avatar
SparkSnail committed
698
                    this.releaseTrialSSHClient(trialJob);
Deshui Yu's avatar
Deshui Yu committed
699
                }
chicm-ms's avatar
chicm-ms committed
700
                this.log.debug(`trailJob status update: ${trialJob.id}, ${trialJob.status}`);
Deshui Yu's avatar
Deshui Yu committed
701
702
703
704
705
706
707
708
709
710
711
712
713
714
            }
            deferred.resolve(trialJob);
        } catch (error) {
            this.log.error(`Update job status exception, error is ${error.message}`);
            if (error instanceof NNIError && error.name === NNIErrorNames.NOT_FOUND) {
                deferred.resolve(trialJob);
            } else {
                trialJob.status = 'UNKNOWN';
                deferred.resolve(trialJob);
            }
        }
        return deferred.promise;
    }

SparkSnail's avatar
SparkSnail committed
715
    private getRemoteScriptsPath(userName: string): string {
716
        return unixPathJoin(getRemoteTmpDir(this.remoteOS), userName, 'nni', 'scripts');
Deshui Yu's avatar
Deshui Yu committed
717
718
719
    }

    private getHostJobRemoteDir(jobId: string): string {
720
        return unixPathJoin(this.remoteExpRootDir, 'hostjobs', jobId);
Deshui Yu's avatar
Deshui Yu committed
721
722
    }

723
    private getRemoteExperimentRootDir(): string{
724
        return unixPathJoin(getRemoteTmpDir(this.remoteOS), 'nni', 'experiments', getExperimentId());
Deshui Yu's avatar
Deshui Yu committed
725
726
    }

SparkSnail's avatar
SparkSnail committed
727
728
729
730
    public get MetricsEmitter() : EventEmitter {
        return this.metricsEmitter;
    }

Deshui Yu's avatar
Deshui Yu committed
731
732
733
734
735
736
737
738
    private getJobPidPath(jobId: string): string {
        const trialJobDetail: RemoteMachineTrialJobDetail | undefined = this.trialJobsMap.get(jobId);
        if (trialJobDetail === undefined) {
            throw new NNIError(NNIErrorNames.INVALID_JOB_DETAIL, `Invalid job detail information for trial job ${jobId}`);
        }

        let jobpidPath: string;
        if (trialJobDetail.form.jobType === 'TRIAL') {
739
            jobpidPath = unixPathJoin(trialJobDetail.workingDirectory, '.nni', 'jobpid');
Deshui Yu's avatar
Deshui Yu committed
740
        } else if (trialJobDetail.form.jobType === 'HOST') {
741
            jobpidPath = unixPathJoin(this.getHostJobRemoteDir(jobId), 'jobpid');
Deshui Yu's avatar
Deshui Yu committed
742
743
744
745
746
747
        } else {
            throw new Error(`Job type not supported: ${trialJobDetail.form.jobType}`);
        }

        return jobpidPath;
    }
chicm-ms's avatar
chicm-ms committed
748
749

    private async writeParameterFile(trialJobId: string, hyperParameters: HyperParameters, rmMeta: RemoteMachineMeta): Promise<void> {
SparkSnail's avatar
SparkSnail committed
750
        const sshClient: Client | undefined = this.trialSSHClientMap.get(trialJobId);
chicm-ms's avatar
chicm-ms committed
751
752
753
754
        if (sshClient === undefined) {
            throw new Error('sshClient is undefined.');
        }

755
        const trialWorkingFolder: string = unixPathJoin(this.remoteExpRootDir, 'trials', trialJobId);
chicm-ms's avatar
chicm-ms committed
756
757
        const trialLocalTempFolder: string = path.join(this.expRootDir, 'trials-local', trialJobId);

758
        const fileName: string = generateParamFileName(hyperParameters);
chicm-ms's avatar
chicm-ms committed
759
760
761
        const localFilepath: string = path.join(trialLocalTempFolder, fileName);
        await fs.promises.writeFile(localFilepath, hyperParameters.value, { encoding: 'utf8' });

762
        await SSHClientUtility.copyFileToRemote(localFilepath, unixPathJoin(trialWorkingFolder, fileName), sshClient);
chicm-ms's avatar
chicm-ms committed
763
    }
764
765

    private generateSequenceId(): number {
766
767
768
769
        if (this.trialSequenceId === -1) {
            this.trialSequenceId = getInitTrialSequenceId();
        }

770
771
        return this.trialSequenceId++;
    }
Deshui Yu's avatar
Deshui Yu committed
772
773
774
}

export { RemoteMachineTrainingService };