1661

pending completion

Build # 1661

Build Type

push

travis-ci-com

Committed by

web-flow

Commit Message

Merge pull request #455 from stanford-oval/wip/mehrad/next

Wip/mehrad/next

Run Details

10846 of 23288 branches covered (46.57%)

Branch coverage included in aggregate %.

13 of 23 new or added lines in 5 files covered. (56.52%)

3 existing lines in 2 files now uncovered.

25401 of 43927 relevant lines covered (57.83%)

541778.36 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

67.44

/lib/prediction/predictor.ts

// -*- mode: typescript; indent-tabs-mode: nil; js-basic-offset: 4 -*-
//
// This file is part of Genie
//
// Copyright 2020 The Board of Trustees of the Leland Stanford Junior University
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//    http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Author: Giovanni Campagna <gcampagn@cs.stanford.edu>

import assert from 'assert';
import * as events from 'events';
import * as child_process from 'child_process';
import * as Tp from 'thingpedia';

import JsonDatagramSocket from '../utils/json_datagram_socket';

const DEFAULT_QUESTION = 'translate from english to thingtalk';

export interface RawPredictionCandidate {
    answer : string;
    score : Record<string, number>;
}

interface Request {
    resolve(data : RawPredictionCandidate[][]) : void;
    reject(err : Error) : void;
}

const DEFAULT_MINIBATCH_SIZE = 30;
const DEFAULT_MAX_LATENCY = 50; // milliseconds

interface Example {
    context : string;
    question : string;
    answer ?: string;
    example_id ?: string;

    resolve(data : RawPredictionCandidate[]) : void;
    reject(err : Error) : void;
}

class LocalWorker extends events.EventEmitter {
    private _child : child_process.ChildProcess|null;
    private _stream : JsonDatagramSocket|null;
    private _nextId : 0;
    private _requests : Map<number, Request>;
    private _modeldir : string;

    constructor(modeldir : string) {
        super();

        this._child = null;
        this._stream = null;
        this._nextId = 0;
        this._requests = new Map;

        this._modeldir = modeldir;
    }

    stop() {
        if (this._child)
            this._child.kill();
        this._child = null;
    }

    start() {
        const args = [
            'server',
            '--stdin',
            '--path', this._modeldir,
        ];
        if (process.env.GENIENLP_EMBEDDINGS)
            args.push('--embeddings', process.env.GENIENLP_EMBEDDINGS);
        if (process.env.GENIENLP_DATABASE_DIR)
            args.push('--database_dir', process.env.GENIENLP_DATABASE_DIR);

        this._child = child_process.spawn('genienlp', args, {
            stdio: ['pipe', 'pipe', 'inherit']
        });
        this._child.on('error', (e) => {
            this._failAll(e);
            this.emit('error', e);
        });
        this._child.on('exit', (code, signal) => {
            //console.error(`Child exited with code ${code}, signal ${signal}`);

            const err = new Error(`Worker died`);
            this._failAll(err);
            this.emit('error', err);
            this._child = null;
            this.emit('exit');
        });

        this._stream = new JsonDatagramSocket(this._child.stdout!, this._child.stdin!, 'utf8');
        this._stream.on('error', (e) => {
            this._failAll(e);
            this.emit('error', e);
        });
        this._stream.on('data', (msg) => {
            const req = this._requests.get(msg.id);
            if (!req) // ignore bogus request
                return;
            if (msg.error) {
                req.reject(new Error(msg.error));
            } else {
                req.resolve(msg.instances.map((instance : any) : RawPredictionCandidate[] => {
                    if (instance.candidates) {
                        return instance.candidates;
                    } else {
                        // no beam search, hence only one candidate
                        // the score might present or not, depending on whether
                        // we calibrate or not
                        return [{
                            answer: instance.answer,
                            score: instance.score || {}
                        }];
                    }
                }));
            }
            this._requests.delete(msg.id);
        });
    }

    private _failAll(error : Error) {
        for (const { reject } of this._requests.values())
            reject(error);
        this._requests.clear();
    }

    request(task : string, minibatch : Example[]) : Promise<RawPredictionCandidate[][]> {
        const id = this._nextId ++;

        return new Promise((resolve, reject) => {
            this._requests.set(id, { resolve, reject });
            //console.error(`${this._requests.size} pending requests`);

            this._stream!.write({ id, task, instances: minibatch }, (err : Error | undefined | null) => {
                if (err) {
                    console.error(err);
                    reject(err);
                }
            });
        });
    }
}

class RemoteWorker extends events.EventEmitter {
    private _url : string;

    constructor(url : string) {
        super();
        this._url = url;
    }

    start() {}
    stop() {}

    async request(task : string, minibatch : Example[]) : Promise<RawPredictionCandidate[][]> {
        const response = await Tp.Helpers.Http.post(this._url, JSON.stringify({
            task,
            instances: minibatch
        }), { dataContentType: 'application/json', accept: 'application/json' });
        return JSON.parse(response).predictions.map((instance : any) : RawPredictionCandidate[] => {
            if (instance.candidates) {
                return instance.candidates;
            } else {
                // no beam search, hence only one candidate
                // the score might present or not, depending on whether
                // we calibrate or not
                return [{
                    answer: instance.answer,
                    score: instance.score || {}
                }];
            }
        });
    }
}

export default class Predictor {
    private _modelurl : string;
    private _worker : LocalWorker|RemoteWorker|null;
    private _stopped : boolean;

    private _minibatchSize : number;
    private _maxLatency : number;

    private _minibatchTask = '';
    private _minibatch : Example[] = [];
    private _minibatchStartTime = 0;

    constructor(modelurl : string, { minibatchSize = DEFAULT_MINIBATCH_SIZE, maxLatency = DEFAULT_MAX_LATENCY }) {
        this._modelurl = modelurl;
        this._worker = null;

        this._minibatchSize = minibatchSize;
        this._maxLatency = maxLatency;

        this._stopped = false;
    }

    private _flushRequest() {
        const minibatch = this._minibatch;
        const task = this._minibatchTask;
        this._minibatch = [];
        this._minibatchTask = '';
        this._minibatchStartTime = 0;

        //console.error(`minibatch: ${minibatch.length} instances`);

        this._worker!.request(task, minibatch).then((candidates) => {
            assert(candidates.length === minibatch.length);
            for (let i = 0; i < minibatch.length; i++)
                minibatch[i].resolve(candidates[i]);
        }, (err : Error) => {
            for (let i = 0; i < minibatch.length; i++)
                minibatch[i].reject(err);
        });
    }

    private _startRequest(ex : Example, task : string, now : number) {
        assert(this._minibatch.length === 0);
        this._minibatch.push(ex);
        this._minibatchTask = task;
        this._minibatchStartTime = now;

        setTimeout(() => {
            if (this._minibatch.length > 0)
                this._flushRequest();
        }, this._maxLatency);
    }

    private _addRequest(ex : Example, task : string) {
        const now = Date.now();
        if (this._minibatch.length === 0) {
            this._startRequest(ex, task, now);
        } else if (this._minibatchTask === task &&
            (now - this._minibatchStartTime < this._maxLatency) &&
            this._minibatch.length < this._minibatchSize) {
            this._minibatch.push(ex);
        } else {
            this._flushRequest();
            this._startRequest(ex, task, now);
        }
    }

    predict(context : string, question = DEFAULT_QUESTION, answer ?: string, task = 'almond', example_id ?: string) : Promise<RawPredictionCandidate[]> {
        assert(typeof context === 'string');
        assert(typeof question === 'string');

        // ensure we have a worker, in case it recently died
        if (!this._worker)
            this.start();

        let resolve ! : (data : RawPredictionCandidate[]) => void,
            reject ! : (err : Error) => void;
        const promise = new Promise<RawPredictionCandidate[]>((_resolve, _reject) => {
            resolve = _resolve;
            reject = _reject;
        });
        this._addRequest({ context, question, answer, resolve, reject }, task);

        return promise;
    }

    start() {
        let worker : RemoteWorker|LocalWorker;
        if (/^kf\+https?:/.test(this._modelurl)) {
            worker = new RemoteWorker(this._modelurl.substring('kf+'.length));
        } else {
            assert(this._modelurl.startsWith('file://'));
            worker = new LocalWorker(this._modelurl.substring('file://'.length));
        }

        worker.on('error', (error : Error) => {
            if (!this._stopped)
                console.error(`Prediction worker had an error: ${error.message}`);
            this._worker = null;

            // fail all the requests in the minibatch if the worker is hosed
            for (const ex of this._minibatch)
                ex.reject(error);
            this._minibatch = [];
            this._minibatchStartTime = 0;
            this._minibatchTask = '';

            worker.stop();
        });
        worker.start();

        this._worker = worker;
    }

    stop() {
        this._stopped = true;
        if (this._worker)
            this._worker.stop();
    }

    reload() {
        // stop the worker, if any
        if (this._worker)
            this._worker.stop();

        // start again
        this.start();
    }
}

1	// -- mode: typescript; indent-tabs-mode: nil; js-basic-offset: 4 --
2	//
3	// This file is part of Genie
4	//
5	// Copyright 2020 The Board of Trustees of the Leland Stanford Junior University
6	//
7	// Licensed under the Apache License, Version 2.0 (the "License");
8	// you may not use this file except in compliance with the License.
9	// You may obtain a copy of the License at
10	//
11	// http://www.apache.org/licenses/LICENSE-2.0
12	//
13	// Unless required by applicable law or agreed to in writing, software
14	// distributed under the License is distributed on an "AS IS" BASIS,
15	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16	// See the License for the specific language governing permissions and
17	// limitations under the License.
18	//
19	// Author: Giovanni Campagna <gcampagn@cs.stanford.edu>
20
21	import assert from 'assert';	61✔
22	import * as events from 'events';	61✔
23	import * as child_process from 'child_process';	61✔
24	import * as Tp from 'thingpedia';	61✔
25
26	import JsonDatagramSocket from '../utils/json_datagram_socket';	61✔
27
28	const DEFAULT_QUESTION = 'translate from english to thingtalk';	61✔
29
30	export interface RawPredictionCandidate {
31	answer : string;
32	score : Record<string, number>;
33	}
34
35	interface Request {
36	resolve(data : RawPredictionCandidate[][]) : void;
37	reject(err : Error) : void;
38	}
39
40	const DEFAULT_MINIBATCH_SIZE = 30;	61✔
41	const DEFAULT_MAX_LATENCY = 50; // milliseconds	61✔
42
43	interface Example {
44	context : string;
45	question : string;
46	answer ?: string;
47	example_id ?: string;
48
49	resolve(data : RawPredictionCandidate[]) : void;
50	reject(err : Error) : void;
51	}
52
53	class LocalWorker extends events.EventEmitter {
54	private _child : child_process.ChildProcess\|null;
55	private _stream : JsonDatagramSocket\|null;
56	private _nextId : 0;
57	private _requests : Map<number, Request>;
58	private _modeldir : string;
59
60	constructor(modeldir : string) {
61	super();	4✔
62
63	this._child = null;	4✔
64	this._stream = null;	4✔
65	this._nextId = 0;	4✔
66	this._requests = new Map;	4✔
67
68	this._modeldir = modeldir;	4✔
69	}
70
71	stop() {
72	if (this._child)	4✔
73	this._child.kill();	2✔
74	this._child = null;	4✔
75	}
76
77	start() {
78	const args = [	4✔
79	'server',
80	'--stdin',
81	'--path', this._modeldir,
82	];
83	if (process.env.GENIENLP_EMBEDDINGS)	4!
84	args.push('--embeddings', process.env.GENIENLP_EMBEDDINGS);	×
85	if (process.env.GENIENLP_DATABASE_DIR)	4!
NEW 86	args.push('--database_dir', process.env.GENIENLP_DATABASE_DIR);	×
87
88	this._child = child_process.spawn('genienlp', args, {	4✔
89	stdio: ['pipe', 'pipe', 'inherit']
90	});
91	this._child.on('error', (e) => {	4✔
92	this._failAll(e);	×
93	this.emit('error', e);	×
94	});
95	this._child.on('exit', (code, signal) => {	4✔
96	//console.error(`Child exited with code ${code}, signal ${signal}`);
97
98	const err = new Error(`Worker died`);	2✔
99	this._failAll(err);	2✔
100	this.emit('error', err);	2✔
101	this._child = null;	2✔
102	this.emit('exit');	2✔
103	});
104
105	this._stream = new JsonDatagramSocket(this._child.stdout!, this._child.stdin!, 'utf8');	4✔
106	this._stream.on('error', (e) => {	4✔
107	this._failAll(e);	×
108	this.emit('error', e);	×
109	});
110	this._stream.on('data', (msg) => {	4✔
111	const req = this._requests.get(msg.id);	2✔
112	if (!req) // ignore bogus request	2!
113	return;	×
114	if (msg.error) {	2!
115	req.reject(new Error(msg.error));	×
116	} else {
117	req.resolve(msg.instances.map((instance : any) : RawPredictionCandidate[] => {	2✔
118	if (instance.candidates) {	2!
119	return instance.candidates;	×
120	} else {
121	// no beam search, hence only one candidate
122	// the score might present or not, depending on whether
123	// we calibrate or not
124	return [{	2✔
125	answer: instance.answer,
126	score: instance.score \|\| {}	4✔
127	}];
128	}
129	}));
130	}
131	this._requests.delete(msg.id);	2✔
132	});
133	}
134
135	private _failAll(error : Error) {
136	for (const { reject } of this._requests.values())	2✔
137	reject(error);	×
138	this._requests.clear();	2✔
139	}
140
141	request(task : string, minibatch : Example[]) : Promise<RawPredictionCandidate[][]> {
142	const id = this._nextId ++;	2✔
143
144	return new Promise((resolve, reject) => {	2✔
145	this._requests.set(id, { resolve, reject });	2✔
146	//console.error(`${this._requests.size} pending requests`);
147
148	this._stream!.write({ id, task, instances: minibatch }, (err : Error \| undefined \| null) => {	2✔
149	if (err) {	2!
150	console.error(err);	×
151	reject(err);	×
152	}
153	});
154	});
155	}
156	}
157
158	class RemoteWorker extends events.EventEmitter {
159	private _url : string;
160
161	constructor(url : string) {
162	super();	×
163	this._url = url;	×
164	}
165
166	start() {}
167	stop() {}
168
169	async request(task : string, minibatch : Example[]) : Promise<RawPredictionCandidate[][]> {
170	const response = await Tp.Helpers.Http.post(this._url, JSON.stringify({	×
171	task,
172	instances: minibatch
173	}), { dataContentType: 'application/json', accept: 'application/json' });
174	return JSON.parse(response).predictions.map((instance : any) : RawPredictionCandidate[] => {	×
175	if (instance.candidates) {	×
176	return instance.candidates;	×
177	} else {
178	// no beam search, hence only one candidate
179	// the score might present or not, depending on whether
180	// we calibrate or not
181	return [{	×
182	answer: instance.answer,
183	score: instance.score \|\| {}	×
184	}];
185	}
186	});
187	}
188	}
189
190	export default class Predictor {	61✔
191	private _modelurl : string;
192	private _worker : LocalWorker\|RemoteWorker\|null;
193	private _stopped : boolean;
194
195	private _minibatchSize : number;
196	private _maxLatency : number;
197
198	private _minibatchTask = '';	4✔
199	private _minibatch : Example[] = [];	4✔
200	private _minibatchStartTime = 0;	4✔
201
202	constructor(modelurl : string, { minibatchSize = DEFAULT_MINIBATCH_SIZE, maxLatency = DEFAULT_MAX_LATENCY }) {	8✔
203	this._modelurl = modelurl;	4✔
204	this._worker = null;	4✔
205
206	this._minibatchSize = minibatchSize;	4✔
207	this._maxLatency = maxLatency;	4✔
208
209	this._stopped = false;	4✔
210	}
211
212	private _flushRequest() {
213	const minibatch = this._minibatch;	2✔
214	const task = this._minibatchTask;	2✔
215	this._minibatch = [];	2✔
216	this._minibatchTask = '';	2✔
217	this._minibatchStartTime = 0;	2✔
218
219	//console.error(`minibatch: ${minibatch.length} instances`);
220
221	this._worker!.request(task, minibatch).then((candidates) => {	2✔
222	assert(candidates.length === minibatch.length);	2✔
223	for (let i = 0; i < minibatch.length; i++)	2✔
224	minibatch[i].resolve(candidates[i]);	2✔
225	}, (err : Error) => {
226	for (let i = 0; i < minibatch.length; i++)	×
227	minibatch[i].reject(err);	×
228	});
229	}
230
231	private _startRequest(ex : Example, task : string, now : number) {
232	assert(this._minibatch.length === 0);	2✔
233	this._minibatch.push(ex);	2✔
234	this._minibatchTask = task;	2✔
235	this._minibatchStartTime = now;	2✔
236
237	setTimeout(() => {	2✔
238	if (this._minibatch.length > 0)	2!
239	this._flushRequest();	2✔
240	}, this._maxLatency);
241	}
242
243	private _addRequest(ex : Example, task : string) {
244	const now = Date.now();	2✔
245	if (this._minibatch.length === 0) {	2!
246	this._startRequest(ex, task, now);	2✔
247	} else if (this._minibatchTask === task &&	×
248	(now - this._minibatchStartTime < this._maxLatency) &&
249	this._minibatch.length < this._minibatchSize) {
250	this._minibatch.push(ex);	×
251	} else {
252	this._flushRequest();	×
253	this._startRequest(ex, task, now);	×
254	}
255	}
256
257	predict(context : string, question = DEFAULT_QUESTION, answer ?: string, task = 'almond', example_id ?: string) : Promise<RawPredictionCandidate[]> {	2!
258	assert(typeof context === 'string');	2✔
259	assert(typeof question === 'string');	2✔
260
261	// ensure we have a worker, in case it recently died
262	if (!this._worker)	2!
263	this.start();	×
264
265	let resolve ! : (data : RawPredictionCandidate[]) => void,
266	reject ! : (err : Error) => void;
267	const promise = new Promise<RawPredictionCandidate[]>((_resolve, _reject) => {	2✔
268	resolve = _resolve;	2✔
269	reject = _reject;	2✔
270	});
271	this._addRequest({ context, question, answer, resolve, reject }, task);	2✔
272
273	return promise;	2✔
274	}
275
276	start() {
277	let worker : RemoteWorker\|LocalWorker;
278	if (/^kf\+https?:/.test(this._modelurl)) {	4!
279	worker = new RemoteWorker(this._modelurl.substring('kf+'.length));	×
280	} else {
281	assert(this._modelurl.startsWith('file://'));	4✔
282	worker = new LocalWorker(this._modelurl.substring('file://'.length));	4✔
283	}
284
285	worker.on('error', (error : Error) => {	4✔
286	if (!this._stopped)	2!
287	console.error(`Prediction worker had an error: ${error.message}`);	×
288	this._worker = null;	2✔
289
290	// fail all the requests in the minibatch if the worker is hosed
291	for (const ex of this._minibatch)	2✔
292	ex.reject(error);	×
293	this._minibatch = [];	2✔
294	this._minibatchStartTime = 0;	2✔
295	this._minibatchTask = '';	2✔
296
297	worker.stop();	2✔
298	});
299	worker.start();	4✔
300
301	this._worker = worker;	4✔
302	}
303
304	stop() {
305	this._stopped = true;	2✔
306	if (this._worker)	2!
307	this._worker.stop();	2✔
308	}
309
310	reload() {
311	// stop the worker, if any
312	if (this._worker)	×
313	this._worker.stop();	×
314
315	// start again
316	this.start();	×
317	}
318	}

stanford-oval / genie-toolkit / 1661

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous