1002

pending completion

Build # 1002

Build Type

Pull #210

travis-ci-com

Committed by

web-flow

Commit Message

Merge a5861865e into 5a48f6aa3

Pull Request Pull Request #210: New round of templates for Thingpedia

Run Details

6532 of 14759 branches covered (44.26%)

Branch coverage included in aggregate %.

178 of 237 new or added lines in 18 files covered. (75.11%)

2361 existing lines in 60 files now uncovered.

12407 of 21760 relevant lines covered (57.02%)

2470310.06 hits per line

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

5.17

/lib/prediction/predictor.js

// -*- mode: js; 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>
"use strict";

const assert = require('assert');
const events = require('events');
const child_process = require('child_process');

const JsonDatagramSocket = require('../utils/json_datagram_socket');

const DEFAULT_QUESTION = 'translate from english to thingtalk';

class Worker extends events.EventEmitter {
    constructor(id, modeldir) {
        super();

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

        this._modeldir = modeldir;
    }

    get ok() {
        return this._child !== null && !this._hadError;
    }

    get busy() {
        return this._requests.size > 0;
    }

    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);

        this._child = child_process.spawn('genienlp', args, {
            stdio: ['pipe', 'pipe', 'inherit']
        });
        this._child.on('error', (e) => {
            this._failAll(e);
            this._hadError = true;
            this.emit('error', e);
        });
        this._child.on('exit', () => {
            this._failAll(new Error(`Worker died`));
            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._hadError = true;
            this.emit('error', e);
        });
        this._stream.on('data', (msg) => {
            if (msg.error) {
                this._requests.get(msg.id).reject(new Error(msg.error));
            } else if (msg.candidates) {
                this._requests.get(msg.id).resolve(msg.candidates);
            } else {
                // no beam search, hence only one candidate, and fixed score
                this._requests.get(msg.id).resolve([{
                    answer: msg.answer,
                    score: 1
                }]);
            }
            this._requests.delete(msg.id);
        });
    }

    _failAll(error) {
        for (let { reject } of this._requests.values())
            reject(error);
        this._requests.clear();
    }

    request(task, context, question) {
        const id = this._nextId ++;

        let resolve, reject;
        const promise = new Promise((_resolve, _reject) => {
            resolve = _resolve;
            reject = _reject;
        });
        this._requests.set(id, { resolve, reject });

        assert(typeof context === 'string');
        assert(typeof question === 'string');
        this._stream.write({ id, context, question, task });
        return promise;
    }
}

module.exports = class Predictor {
    constructor(id, modeldir, nWorkers) {
        this._nWorkers = nWorkers;

        this.id = id;
        this._modeldir = modeldir;
        this._nextId = 0;
        this._workers = new Set;

        this._stopped = false;
    }

    start() {
        //console.log(`Spawning ${this._nWorkers} workers for predictor ${this.id}`);
        for (let i = 0; i < this._nWorkers; i++)
            this._startWorker();
    }

    stop() {
        this._stopped = true;
        this._killall();
    }

    _killall() {
        for (let worker of this._workers)
            worker.stop();
    }

    reload() {
        // stop all workers and clear them up
        this._killall();
        this._workers.clear();

        // start again
        this.start();
    }

    _startWorker() {
        const worker = new Worker(`${this.id}/${this._nextId++}`, this._modeldir);
        worker.on('error', (err) => {
            console.error(`Worker ${worker.id} had an error: ${err.message}`);
            worker.stop();
        });
        worker.on('exit', (err) => {
            if (!this._stopped)
                console.error(`Worker ${worker.id} exited`);
            this._workers.delete(worker);

            if (!this._stopped) {
                // wait 30 seconds, then autorespawn the worker
                // this ensures that we don't stay with fewer workers than
                // we should for too long, as that can overload the few workers
                // who are alive, and cause latency issues
                setTimeout(() => {
                    if (this._workers.size < this._nWorkers)
                        this._startWorker();
                }, 30000);
            }
        });

        worker.start();
        this._workers.add(worker);
        return worker;
    }

    predict(context, question = DEFAULT_QUESTION, task = 'almond') {
        // first pick a worker that is free
        for (let worker of this._workers) {
            if (worker.ok && !worker.busy)
                return worker.request(task, context, question);
        }

        // failing that, pick any worker that is alive
        for (let worker of this._workers) {
            if (worker.ok)
                return worker.request(task, context, question);
        }

        // failing that, spawn a new worker
        return this._startWorker().request(task, context, question);
    }
};

1	// -- mode: js; 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	"use strict";
21
22	const assert = require('assert');	29✔
23	const events = require('events');	29✔
24	const child_process = require('child_process');	29✔
25
26	const JsonDatagramSocket = require('../utils/json_datagram_socket');	29✔
27
28	const DEFAULT_QUESTION = 'translate from english to thingtalk';	29✔
29
30	class Worker extends events.EventEmitter {
31	constructor(id, modeldir) {
UNCOV 32	super();	×
33
UNCOV 34	this.id = id;	×
UNCOV 35	this._child = null;	×
UNCOV 36	this._hadError = false;	×
UNCOV 37	this._stream = null;	×
UNCOV 38	this._nextId = 0;	×
UNCOV 39	this._requests = new Map;	×
40
UNCOV 41	this._modeldir = modeldir;	×
42	}
43
44	get ok() {
45	return this._child !== null && !this._hadError;	×
46	}
47
48	get busy() {
49	return this._requests.size > 0;	×
50	}
51
52	stop() {
UNCOV 53	if (this._child)	×
UNCOV 54	this._child.kill();	×
UNCOV 55	this._child = null;	×
56	}
57
58	start() {
UNCOV 59	const args = [	×
60	'server',
61	'--stdin',
62	'--path', this._modeldir,
63	];
UNCOV 64	if (process.env.GENIENLP_EMBEDDINGS)	×
65	args.push('--embeddings', process.env.GENIENLP_EMBEDDINGS);	×
66
UNCOV 67	this._child = child_process.spawn('genienlp', args, {	×
68	stdio: ['pipe', 'pipe', 'inherit']
69	});
UNCOV 70	this._child.on('error', (e) => {	×
71	this._failAll(e);	×
72	this._hadError = true;	×
73	this.emit('error', e);	×
74	});
UNCOV 75	this._child.on('exit', () => {	×
76	this._failAll(new Error(`Worker died`));	×
77	this._child = null;	×
78	this.emit('exit');	×
79	});
80
UNCOV 81	this._stream = new JsonDatagramSocket(this._child.stdout, this._child.stdin, 'utf8');	×
82
UNCOV 83	this._stream.on('error', (e) => {	×
84	this._failAll(e);	×
85	this._hadError = true;	×
86	this.emit('error', e);	×
87	});
UNCOV 88	this._stream.on('data', (msg) => {	×
UNCOV 89	if (msg.error) {	×
90	this._requests.get(msg.id).reject(new Error(msg.error));	×
UNCOV 91	} else if (msg.candidates) {	×
92	this._requests.get(msg.id).resolve(msg.candidates);	×
93	} else {
94	// no beam search, hence only one candidate, and fixed score
UNCOV 95	this._requests.get(msg.id).resolve([{	×
96	answer: msg.answer,
97	score: 1
98	}]);
99	}
UNCOV 100	this._requests.delete(msg.id);	×
101	});
102	}
103
104	_failAll(error) {
105	for (let { reject } of this._requests.values())	×
106	reject(error);	×
107	this._requests.clear();	×
108	}
109
110	request(task, context, question) {
UNCOV 111	const id = this._nextId ++;	×
112
113	let resolve, reject;
UNCOV 114	const promise = new Promise((_resolve, _reject) => {	×
UNCOV 115	resolve = _resolve;	×
UNCOV 116	reject = _reject;	×
117	});
UNCOV 118	this._requests.set(id, { resolve, reject });	×
119
UNCOV 120	assert(typeof context === 'string');	×
UNCOV 121	assert(typeof question === 'string');	×
UNCOV 122	this._stream.write({ id, context, question, task });	×
UNCOV 123	return promise;	×
124	}
125	}
126
127	module.exports = class Predictor {	29✔
128	constructor(id, modeldir, nWorkers) {
UNCOV 129	this._nWorkers = nWorkers;	×
130
UNCOV 131	this.id = id;	×
UNCOV 132	this._modeldir = modeldir;	×
UNCOV 133	this._nextId = 0;	×
UNCOV 134	this._workers = new Set;	×
135
UNCOV 136	this._stopped = false;	×
137	}
138
139	start() {
140	//console.log(`Spawning ${this._nWorkers} workers for predictor ${this.id}`);
UNCOV 141	for (let i = 0; i < this._nWorkers; i++)	×
142	this._startWorker();	×
143	}
144
145	stop() {
UNCOV 146	this._stopped = true;	×
UNCOV 147	this._killall();	×
148	}
149
150	_killall() {
UNCOV 151	for (let worker of this._workers)	×
UNCOV 152	worker.stop();	×
153	}
154
155	reload() {
156	// stop all workers and clear them up
157	this._killall();	×
158	this._workers.clear();	×
159
160	// start again
161	this.start();	×
162	}
163
164	_startWorker() {
UNCOV 165	const worker = new Worker(`${this.id}/${this._nextId++}`, this._modeldir);	×
UNCOV 166	worker.on('error', (err) => {	×
167	console.error(`Worker ${worker.id} had an error: ${err.message}`);	×
168	worker.stop();	×
169	});
UNCOV 170	worker.on('exit', (err) => {	×
171	if (!this._stopped)	×
172	console.error(`Worker ${worker.id} exited`);	×
173	this._workers.delete(worker);	×
174
175	if (!this._stopped) {	×
176	// wait 30 seconds, then autorespawn the worker
177	// this ensures that we don't stay with fewer workers than
178	// we should for too long, as that can overload the few workers
179	// who are alive, and cause latency issues
180	setTimeout(() => {	×
181	if (this._workers.size < this._nWorkers)	×
182	this._startWorker();	×
183	}, 30000);
184	}
185	});
186
UNCOV 187	worker.start();	×
UNCOV 188	this._workers.add(worker);	×
UNCOV 189	return worker;	×
190	}
191
192	predict(context, question = DEFAULT_QUESTION, task = 'almond') {	×
193	// first pick a worker that is free
UNCOV 194	for (let worker of this._workers) {	×
195	if (worker.ok && !worker.busy)	×
196	return worker.request(task, context, question);	×
197	}
198
199	// failing that, pick any worker that is alive
UNCOV 200	for (let worker of this._workers) {	×
201	if (worker.ok)	×
202	return worker.request(task, context, question);	×
203	}
204
205	// failing that, spawn a new worker
UNCOV 206	return this._startWorker().request(task, context, question);	×
207	}
208	};

stanford-oval / genie-toolkit / 1002

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