FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

Use worker pool to get environment info (#13386) · Barada01/pythonVSCode@01fc17d · GitHub

/ pythonVSCode Public
forked from DonJayamanne/pythonVSCode

Commit 01fc17d

Browse files
karthiknadig
authored
Use worker pool to get environment info (#13386)
* Generic worker pool * Add environment info service * Add more tests * Rebase with master * Clean up * Address comments * Add cache to environment info calls * Rebase with master and some more cleanup * Register class with ioc. * Address comments. * Move workerpool to utils * Addressed more comments * Typos and clean up. * Fix build
1 parent cf074f4 commit 01fc17d

19 files changed

Lines changed: 614 additions & 18 deletions

‎package-lock.json‎

Lines changed: 6 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎package.json‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3761,6 +3761,7 @@
37613761
"thread-loader": "^2.1.3",
37623762
"transform-loader": "^0.2.4",
37633763
"ts-loader": "^5.3.0",
3764+
"ts-mock-imports": "^1.3.0",
37643765
"ts-mockito": "^2.5.0",
37653766
"ts-node": "^8.3.0",
37663767
"tsconfig-paths-webpack-plugin": "^3.2.0",

‎src/client/common/utils/workerPool.ts‎

Lines changed: 250 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,250 @@
1+
// Copyright (c) Microsoft Corporation. All rights reserved.
2+
// Licensed under the MIT License.
3+
4+
import { traceError } from '../logger';
5+
import { createDeferred, Deferred } from './async';
6+
7+
interface IWorker {
8+
/**
9+
* Start processing of items.
10+
* @method stop
11+
*/
12+
start(): void;
13+
/**
14+
* Stops any further processing of items.
15+
* @method stop
16+
*/
17+
stop(): void;
18+
}
19+
20+
type NextFunc<T> = () => Promise<T>;
21+
type WorkFunc<T, R> = (item: T) => Promise<R>;
22+
type PostResult<T, R> = (item: T, result?: R, err?: Error) => void;
23+
24+
interface IWorkItem<T> {
25+
item: T;
26+
}
27+
28+
export enum QueuePosition {
29+
Back,
30+
Front
31+
}
32+
33+
export interface IWorkerPool<T, R> extends IWorker {
34+
/**
35+
* Add items to be processed to a queue.
36+
* @method addToQueue
37+
* @param {T} item: Item to process
38+
* @param {QueuePosition} position: Add items to the front or back of the queue.
39+
* @returns A promise that when resolved gets the result from running the worker function.
40+
*/
41+
addToQueue(item: T, position?: QueuePosition): Promise<R>;
42+
}
43+
44+
class Worker<T, R> implements IWorker {
45+
private stopProcessing: boolean = false;
46+
public constructor(
47+
private readonly next: NextFunc<T>,
48+
private readonly workFunc: WorkFunc<T, R>,
49+
private readonly postResult: PostResult<T, R>,
50+
private readonly name: string
51+
) {}
52+
public stop() {
53+
this.stopProcessing = true;
54+
}
55+
56+
public async start() {
57+
while (!this.stopProcessing) {
58+
try {
59+
const workItem = await this.next();
60+
try {
61+
const result = await this.workFunc(workItem);
62+
this.postResult(workItem, result);
63+
} catch (ex) {
64+
this.postResult(workItem, undefined, ex);
65+
}
66+
} catch (ex) {
67+
// Next got rejected. Likely worker pool is shutting down.
68+
// continue here and worker will exit if the worker pool is shutting down.
69+
traceError(`Error while running worker[${this.name}].`, ex);
70+
continue;
71+
}
72+
}
73+
}
74+
}
75+
76+
class WorkQueue<T, R> {
77+
private readonly items: IWorkItem<T>[] = [];
78+
private readonly results: Map<IWorkItem<T>, Deferred<R>> = new Map();
79+
public add(item: T, position?: QueuePosition): Promise<R> {
80+
// Wrap the user provided item in a wrapper object. This will allow us to track multiple
81+
// submissions of the same item. For example, addToQueue(2), addToQueue(2). If we did not
82+
// wrap this, then from the map both submissions will look the same. Since this is a generic
83+
// worker pool, we do not know if we can resolve both using the same promise. So, a better
84+
// approach is to ensure each gets a unique promise, and let the worker function figure out
85+
// how to handle repeat submissions.
86+
const workItem: IWorkItem<T> = { item };
87+
if (position === QueuePosition.Front) {
88+
this.items.unshift(workItem);
89+
} else {
90+
this.items.push(workItem);
91+
}
92+
93+
// This is the promise that will be resolved when the work
94+
// item is complete. We save this in a map to resolve when
95+
// the worker finishes and posts the result.
96+
const deferred = createDeferred<R>();
97+
this.results.set(workItem, deferred);
98+
99+
return deferred.promise;
100+
}
101+
102+
public completed(workItem: IWorkItem<T>, result?: R, error?: Error): void {
103+
const deferred = this.results.get(workItem);
104+
if (deferred !== undefined) {
105+
this.results.delete(workItem);
106+
if (error !== undefined) {
107+
deferred.reject(error);
108+
}
109+
deferred.resolve(result);
110+
}
111+
}
112+
113+
public next(): IWorkItem<T> | undefined {
114+
return this.items.shift();
115+
}
116+
117+
public clear(): void {
118+
this.results.forEach((v: Deferred<R>, k: IWorkItem<T>, map: Map<IWorkItem<T>, Deferred<R>>) => {
119+
v.reject(Error('Queue stopped processing'));
120+
map.delete(k);
121+
});
122+
}
123+
}
124+
125+
class WorkerPool<T, R> implements IWorkerPool<T, R> {
126+
// This collection tracks the full set of workers.
127+
private readonly workers: IWorker[] = [];
128+
129+
// A collections that holds unblock callback for each worker waiting
130+
// for a work item when the queue is empty
131+
private readonly waitingWorkersUnblockQueue: { unblock(w: IWorkItem<T>): void; stop(): void }[] = [];
132+
133+
// A collection that manages the work items.
134+
private readonly queue = new WorkQueue<T, R>();
135+
136+
// State of the pool manages via stop(), start()
137+
private stopProcessing = false;
138+
139+
public constructor(
140+
private readonly workerFunc: WorkFunc<T, R>,
141+
private readonly numWorkers: number = 2,
142+
private readonly name: string = 'Worker'
143+
) {}
144+
145+
public addToQueue(item: T, position?: QueuePosition): Promise<R> {
146+
if (this.stopProcessing) {
147+
throw Error('Queue is stopped');
148+
}
149+
150+
// This promise when resolved should return the processed result of the item
151+
// being added to the queue.
152+
const deferred = this.queue.add(item, position);
153+
154+
const worker = this.waitingWorkersUnblockQueue.shift();
155+
if (worker) {
156+
const workItem = this.queue.next();
157+
if (workItem !== undefined) {
158+
// If we are here it means there were no items to process in the queue.
159+
// At least one worker is free and waiting for a work item. Call 'unblock'
160+
// and give the worker the newly added item.
161+
worker.unblock(workItem);
162+
} else {
163+
// Something is wrong, we should not be here. we just added an item to
164+
// the queue. It should not be empty.
165+
traceError('Work queue was empty immediately after adding item.');
166+
}
167+
}
168+
169+
return deferred;
170+
}
171+
172+
public start() {
173+
this.stopProcessing = false;
174+
let num = this.numWorkers;
175+
while (num > 0) {
176+
this.workers.push(
177+
new Worker<IWorkItem<T>, R>(
178+
() => this.nextWorkItem(),
179+
(workItem: IWorkItem<T>) => this.workerFunc(workItem.item),
180+
(workItem: IWorkItem<T>, result?: R, error?: Error) =>
181+
this.queue.completed(workItem, result, error),
182+
`${this.name} ${num}`
183+
)
184+
);
185+
num = num - 1;
186+
}
187+
this.workers.forEach(async (w) => w.start());
188+
}
189+
190+
public stop(): void {
191+
this.stopProcessing = true;
192+
193+
// Signal all registered workers with this worker pool to stop processing.
194+
// Workers should complete the task they are currently doing.
195+
let worker = this.workers.shift();
196+
while (worker) {
197+
worker.stop();
198+
worker = this.workers.shift();
199+
}
200+
201+
// Remove items from queue.
202+
this.queue.clear();
203+
204+
// This is necessary to exit any worker that is waiting for an item.
205+
// If we don't unblock here then the worker just remains blocked
206+
// forever.
207+
let blockedWorker = this.waitingWorkersUnblockQueue.shift();
208+
while (blockedWorker) {
209+
blockedWorker.stop();
210+
blockedWorker = this.waitingWorkersUnblockQueue.shift();
211+
}
212+
}
213+
214+
public nextWorkItem(): Promise<IWorkItem<T>> {
215+
// Note that next() will return `undefined` if the queue is empty.
216+
const nextWorkItem = this.queue.next();
217+
if (nextWorkItem !== undefined) {
218+
return Promise.resolve(nextWorkItem);
219+
}
220+
221+
// Queue is Empty, so return a promise that will be resolved when
222+
// new items are added to the queue.
223+
return new Promise<IWorkItem<T>>((resolve, reject) => {
224+
this.waitingWorkersUnblockQueue.push({
225+
unblock: (workItem?: IWorkItem<T>) => {
226+
// This will be called to unblock any worker waiting for items.
227+
if (this.stopProcessing) {
228+
// We should reject here since the processing should be stopped.
229+
reject();
230+
}
231+
// If we are here, the queue received a new work item. Resolve with that item.
232+
resolve(workItem);
233+
},
234+
stop: () => {
235+
reject();
236+
}
237+
});
238+
});
239+
}
240+
}
241+
242+
export function createWorkerPool<T, R>(
243+
workerFunc: WorkFunc<T, R>,
244+
numWorkers: number = 2,
245+
name: string = 'Worker'
246+
): IWorkerPool<T, R> {
247+
const pool = new WorkerPool<T, R>(workerFunc, numWorkers, name);
248+
pool.start();
249+
return pool;
250+
}

‎src/client/extensionInit.ts‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -37,9 +37,9 @@ export function initializeGlobals(context: IExtensionContext): [IServiceManager,
3737

3838
export function initializeComponents(
3939
_context: IExtensionContext,
40-
_serviceManager: IServiceManager,
41-
_serviceContainer: IServiceContainer
40+
serviceManager: IServiceManager,
41+
serviceContainer: IServiceContainer
4242
) {
43-
registerForIOC(_serviceManager);
43+
registerForIOC(serviceManager, serviceContainer);
4444
// We will be pulling code over from activateLegacy().
4545
}

‎src/client/pythonEnvironments/common/externalDependencies.ts‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
// Copyright (c) Microsoft Corporation. All rights reserved.
2+
// Licensed under the MIT License.
3+
4+
import { ExecutionResult, IProcessServiceFactory } from '../../common/process/types';
5+
import { IServiceContainer } from '../../ioc/types';
6+
7+
let internalServiceContainer: IServiceContainer;
8+
export function initializeExternalDependencies(serviceContainer: IServiceContainer): void {
9+
internalServiceContainer = serviceContainer;
10+
}
11+
12+
function getProcessFactory(): IProcessServiceFactory {
13+
return internalServiceContainer.get<IProcessServiceFactory>(IProcessServiceFactory);
14+
}
15+
16+
export async function shellExecute(command: string, timeout: number): Promise<ExecutionResult<string>> {
17+
const proc = await getProcessFactory().create();
18+
return proc.shellExec(command, { timeout });
19+
}

‎src/client/pythonEnvironments/info/environmentInfoService.ts‎

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,85 @@
1+
// Copyright (c) Microsoft Corporation. All rights reserved.
2+
// Licensed under the MIT License.
3+
4+
import { injectable } from 'inversify';
5+
import { EnvironmentType, PythonEnvironment } from '.';
6+
import { createWorkerPool, IWorkerPool, QueuePosition } from '../../common/utils/workerPool';
7+
import { shellExecute } from '../common/externalDependencies';
8+
import { buildPythonExecInfo } from '../exec';
9+
import { getInterpreterInfo } from './interpreter';
10+
11+
export enum EnvironmentInfoServiceQueuePriority {
12+
Default,
13+
High
14+
}
15+
16+
export const IEnvironmentInfoService = Symbol('IEnvironmentInfoService');
17+
export interface IEnvironmentInfoService {
18+
getEnvironmentInfo(
19+
interpreterPath: string,
20+
priority?: EnvironmentInfoServiceQueuePriority
21+
): Promise<PythonEnvironment | undefined>;
22+
}
23+
24+
async function buildEnvironmentInfo(interpreterPath: string): Promise<PythonEnvironment | undefined> {
25+
const interpreterInfo = await getInterpreterInfo(buildPythonExecInfo(interpreterPath), shellExecute);
26+
if (interpreterInfo === undefined || interpreterInfo.version === undefined) {
27+
return undefined;
28+
}
29+
return {
30+
path: interpreterInfo.path,
31+
// Have to do this because the type returned by getInterpreterInfo is SemVer
32+
// But we expect this to be PythonVersion
33+
version: {
34+
raw: interpreterInfo.version.raw,
35+
major: interpreterInfo.version.major,
36+
minor: interpreterInfo.version.minor,
37+
patch: interpreterInfo.version.patch,
38+
build: interpreterInfo.version.build,
39+
prerelease: interpreterInfo.version.prerelease
40+
},
41+
sysVersion: interpreterInfo.sysVersion,
42+
architecture: interpreterInfo.architecture,
43+
sysPrefix: interpreterInfo.sysPrefix,
44+
pipEnvWorkspaceFolder: interpreterInfo.pipEnvWorkspaceFolder,
45+
companyDisplayName: '',
46+
displayName: '',
47+
envType: EnvironmentType.Unknown, // Code to handle This will be added later.
48+
envName: '',
49+
envPath: '',
50+
cachedEntry: false
51+
};
52+
}
53+
54+
@injectable()
55+
export class EnvironmentInfoService implements IEnvironmentInfoService {
56+
// Caching environment here in-memory. This is so that we don't have to run this on the same
57+
// path again and again in a given session. This information will likely not change in a given
58+
// session. There are definitely cases where this will change. But a simple reload should address
59+
// those.
60+
private readonly cache: Map<string, PythonEnvironment> = new Map<string, PythonEnvironment>();
61+
private readonly workerPool: IWorkerPool<string, PythonEnvironment | undefined>;
62+
public constructor() {
63+
this.workerPool = createWorkerPool<string, PythonEnvironment | undefined>(buildEnvironmentInfo);
64+
}
65+
66+
public async getEnvironmentInfo(
67+
interpreterPath: string,
68+
priority?: EnvironmentInfoServiceQueuePriority
69+
): Promise<PythonEnvironment | undefined> {
70+
const result = this.cache.get(interpreterPath);
71+
if (result !== undefined) {
72+
return result;
73+
}
74+
75+
return (priority === EnvironmentInfoServiceQueuePriority.High
76+
? this.workerPool.addToQueue(interpreterPath, QueuePosition.Front)
77+
: this.workerPool.addToQueue(interpreterPath, QueuePosition.Back)
78+
).then((r) => {
79+
if (r !== undefined) {
80+
this.cache.set(interpreterPath, r);
81+
}
82+
return r;
83+
});
84+
}
85+
}

‎src/client/pythonEnvironments/info/index.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,7 @@ export type PythonVersionInfo = [number, number, number, ReleaseLevel];
4848
export type InterpreterInformation = {
4949
path: string;
5050
version?: PythonVersion;
51-
sysVersion: string;
51+
sysVersion?: string;
5252
architecture: Architecture;
5353
sysPrefix: string;
5454
pipEnvWorkspaceFolder?: string;

0 commit comments

Comments
 (0)
Back | FazBrowse Home | New Git URL