diff --git a/app.js b/app.js index 6eb4bdab..408cea4e 100644 --- a/app.js +++ b/app.js @@ -446,7 +446,7 @@ const script = { "load/file", "moto/broker", "moto/idb", - "main/mesh" + "main/mesh-ui" ], mesh_work : [ "moto/worker", @@ -468,7 +468,8 @@ const script = { "geo/polygons", "geo/mesh", "moto/broker", - "moto/idb" + "moto/idb", + "main/mesh-work" ] }; diff --git a/src/main/mesh.js b/src/main/mesh-ui.js similarity index 96% rename from src/main/mesh.js rename to src/main/mesh-ui.js index 12800eb3..41102e7d 100644 --- a/src/main/mesh.js +++ b/src/main/mesh-ui.js @@ -21,10 +21,10 @@ let DOC = document, function init() { let moto = self.moto, - sky = false; + sky = false, dark = false, ortho = false, - zoomrev = true; + zoomrev = true, zoomspd = 1, space = moto.Space, platform = space.platform; diff --git a/src/main/mesh-work.js b/src/main/mesh-work.js new file mode 100644 index 00000000..4509d061 --- /dev/null +++ b/src/main/mesh-work.js @@ -0,0 +1,7 @@ +/** Copyright Stewart Allen -- All Rights Reserved */ + +"use strict"; + +(function() { + +})(); \ No newline at end of file diff --git a/src/moto/client.js b/src/moto/client.js index c2c49253..b15c79f9 100644 --- a/src/moto/client.js +++ b/src/moto/client.js @@ -6,47 +6,111 @@ let gapp = self.gapp = self.gapp || {}, moto = self.moto = self.moto || {}, - time = function() { return new Date().getTime() }, - restarting = false, - running = {}, - workurl = null, - worker = null, - seqid = 1; + ccvalue = self.navigator ? navigator.hardwareConcurrency || 1 : 1, + ccmax = ccvalue > 3 ? ccvalue - 1 : 0, + workcc = false, // concurrent or not + workurl = null, // url to load worker + worknum = 1, // number of workers to start + workers = [], // array of all workers + queue = []; // array of work requests if (moto.client) return; -/** - * @param {Function} fn name of function in moto.worker - * @param {Object} data to send to server - * @param {Function} onreply function to call on reply messages - * @param {Object[]} zerocopy array of objects to pass using zerocopy - */ -function send(fn, data, onreply, zerocopy) { - let seq = seqid++; - - if (onreply) { - // establish listener and track replies - running[seq] = { fn:onreply }; - } - let msg = { - seq: seq, - task: fn, - time: time(), - data: data - }; - try { - worker.postMessage(msg, zerocopy); - } catch (error) { - console.trace('work send error', {data, error}); - } -} - // code is running in the browser / client context let client = moto.client = { - send: send, + + active: () => { + return workers.length; + }, + + call: (task, data, options = {}) => { + let { direct, stream } = options; + return new Promise((resolve, reject) => { + let ondata, ondone, onerror = reject; + if (!stream) { + let accum = []; + ondata = (data) => { + accum.push(data); + }; + ondone = () => { + resolve(accum); + }; + } else { + ondata = stream; + ondone = resolve; + } + client.queue(task, data, { direct, ondata, ondone, onerror }); + }); + }, + + queue: (task, data, options = {}) => { + queue.push({task, data, options}); + client.kick(); + }, + + kick: () => { + if (queue.length === 0) { + return; + } + + let rec, worker; + outer: for (let qrec of queue) { + let { options } = qrec; + let { index } = options; + if (index >= 0) { + // worker designated by index + let mod = index % workers.length; + if (workers[mod].run === null) { + worker = workers[mod]; + rec = qrec; + break; + } + } else { + // first available worker + for (let i=0; i w.run).length; }, stop: function() { - if (worker) { - worker.terminate(); - worker = null; + for (let w of workers) { + w.terminate(); } + workers = []; }, // same as restart - start: function(url) { - client.restart(url); + start: function(url = workurl, cc = workcc) { + workcc = cc; + worknum = cc ? ccmax : 1; + workurl = url; + client.restart(url, cc); }, - restart: function(url) { - // prevent re-entry from cancel callback - if (restarting) { - return; - } - + restart: function() { client.stop(); - restarting = true; - - for (let key in running) { - let rec = running[key]; - if (rec.fn) { - rec.fn({error: "cancelled operation"}); + // send cancel/error notification to requestors + for (let w of workers.filter(w => w.run)) { + if (w.run.ondone) { + w.run.ondone({error: "cancelled operation"}); } } - running = {}; - worker = client.newWorker(url); + for (let i=0; i { endpoints[name] = fn; - sync(name); }, send: (msg) => { self.postMessage(msg); }, - sync: (name) => { - if (name) { - if (isinit) dispatch.send({register: [name]}); - } else { - dispatch.send({register: Object.keys(endpoints)}); - } - }, - onmessage: (e) => { - - if (e.data === '--init--') { - isinit = true; - dispatch.sync(); - return; - } - - let time_recv = time(), - msg = e.data || {}, + let msg = e.data, run = endpoints[msg.task], + done = false, send = { - data : function(data,direct) { - dispatch.send({ - seq: msg.seq, - task: msg.task, - done: false, - data: data - }, direct); + data: (data, direct) => { + if (canSend(data)) { + dispatch.send({ done: false, data: data }, direct); + } }, - done : function(data,direct) { - dispatch.send({ - seq: msg.seq, - task: msg.task, - done: true, - data: data - }, direct); + done: (data, direct) => { + if (canSend(data)) { + dispatch.send({ done: true, data: data }, direct); + done = true; + } + }, + error: (data) => { + dispatch.send({ error: data }); } + }, + canSend = (data) => { + if (done) { + console.log("unexpected reply dropped", {data, msg}); + } + return !done; }; if (run) { try { - let time_xfer = (time_recv - msg.time), - output = run(msg.data, send), - time_send = time(), - time_proc = time_send - time_recv; - - if (output) dispatch.send({ - seq: msg.seq, - task: msg.task, - time_send: time_xfer, - time_proc: time_proc, - // replaced on reply side - time_recv: time(), - data: output - }); - } catch (wrkerr) { - // console.log(wrkerr); - console.trace(wrkerr.stack); - send.done({error: wrkerr.toString()}); + let data = run(msg.data, send); + if (data) { + send.done(data); + } + } catch (error) { + console.trace(error.stack); + dispatch.send({ data: {error: error.toString()}, done: true }); } } else { - console.log({worker_unhandled: e, msg, fn: dispatch[msg.task]}); + console.log({worker_unhandled: e}); + send.error(`no registered endpoint: ${msg.task}`); } }