split mesh ui/work, change client/worker contract

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"
     ]
 };
 

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;

src/main/mesh-work.js

@@ -0,0 +1,7 @@
+/** Copyright Stewart Allen <sa@grid.space> -- All Rights Reserved */
+
+"use strict";
+
+(function() {
+
+})();

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() },
+    ccvalue = self.navigator ? navigator.hardwareConcurrency || 1 : 1,
-    restarting = false,
+    ccmax = ccvalue > 3 ? ccvalue - 1 : 0,
-    running = {},
+    workcc = false, // concurrent or not
-    workurl = null,
+    workurl = null, // url to load worker
-    worker = null,
+    worknum = 1,    // number of workers to start
-    seqid = 1;
+    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<workers.length; i++) {
+                    if (workers[i].run === null) {
+                        worker = workers[i];
+                        rec = qrec;
+                        break outer;
+                    }
+                }
+            }
+        }
+
+        if (!rec) {
+            // nothing runnable
+            return;
+        }
+
+        // remove queue item
+        let qindex = queue.indexOf(rec);
+        if (qindex < 0) {
+            throw "invalid queue record";
+        }
+        queue.splice(qindex,1);
+
+        let { task, data, options } = rec;
+        let { direct, ondone, ondata, onerror } = options;
+
+        if (ondone || ondata) {
+            worker.run = { ondone, ondata, onerror };
+        }
+
+        let msg = {
+            task,
+            data,
+            stream: ondata ? true : false
+        };
+
+        try {
+            worker.postMessage(msg, direct);
+        } catch (error) {
+            console.trace('work send error', {data, error});
+        }
+    },
 
     // factory can be overridden
-    newWorker: function(url = workurl) {
+    spawn: function(url = workurl) {
         if (self.createWorker) {
             return self.createWorker();
         } else {
@@ -55,76 +119,80 @@ let client = moto.client = {
     },
 
     // return number of running requests
-    current: function() {
+    running: function() {
-        let current = 0;
+        return workers.filter(w => w.run).length;
-        for (let rec of Object.values(running)) {
-            if (rec.fn) current++;
-        }
-        return current;
     },
 
     stop: function() {
-        if (worker) {
+        for (let w of workers) {
-            worker.terminate();
+            w.terminate();
-            worker = null;
         }
+        workers = [];
     },
 
     // same as restart
-    start: function(url) {
+    start: function(url = workurl, cc = workcc) {
-        client.restart(url);
+        workcc = cc;
+        worknum = cc ? ccmax : 1;
+        workurl = url;
+        client.restart(url, cc);
     },
 
-    restart: function(url) {
+    restart: function() {
-        // prevent re-entry from cancel callback
-        if (restarting) {
-            return;
-        }
-
         client.stop();
 
-        restarting = true;
+        // send cancel/error notification to requestors
-
+        for (let w of workers.filter(w => w.run)) {
-        for (let key in running) {
+            if (w.run.ondone) {
-            let rec = running[key];
+                w.run.ondone({error: "cancelled operation"});
-            if (rec.fn) {
-                rec.fn({error: "cancelled operation"});
             }
         }
 
-        running = {};
+        for (let i=0; i<worknum; i++) {
-        worker = client.newWorker(url);
+            let worker = client.spawn(workurl);
 
-        // bind to client to enable synth injections
+            worker.run = null;
-        client.onmessage = worker.onmessage = function(e) {
+            worker.index = i;
-            let now = time(),
+            worker.onmessage = function(e) {
-                reply = e.data,
+                if (worker.run === null) {
-                record = running[reply.seq],
+                    console.log({unhandled: e});
-                onreply = record ? record.fn : undefined;
+                    throw "message on idle worker";
+                }
 
-            if (reply.register) {
+                let reply = e.data;
-                // console.log({register: reply.register});
+                let { done, data, error } = reply;
-                return;
+                let { ondone, ondata, onerror } = worker.run;
-            }
+                let handled = false;
 
-            if (reply.done) {
+                if (error) {
-                delete running[reply.seq];
+                    onerror(error);
-            }
+                    worker.run = null;
+                    handled = true;
+                    client.kick();
+                }
 
-            // calculate and replace recv time
+                if (ondata && data) {
-            reply.time_recv = now - reply.time_recv;
+                    ondata(data);
+                    handled = true;
+                }
 
-            if (onreply) {
+                if (done) {
-                onreply(reply.data, reply);
+                    ondone();
-            } else {
+                    worker.run = null;
-                console.log({drop_reply: reply});
+                    handled = true;
-            }
+                    client.kick();
-        };
+                }
 
-        // signal worker client is ready
+                if (!handled) {
-        worker.postMessage("--init--");
+                    console.log({unhandled: e});
+                    throw "unhandled reply";
+                }
+            };
 
-        restarting = false;
+            workers.push(worker);
+        }
+
+        client.kick();
     }
 
 };

src/moto/worker.js

@@ -13,7 +13,6 @@ if (moto.worker) return;
 // allow license to inject module
 self.gapp = self.gapp || moto;
 
-let isinit = false;
 let endpoints = {};
 
 // code is running in the worker / server context
@@ -21,74 +20,52 @@ const dispatch = moto.worker = {
 
     register: (name, fn) => {
         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) => {
-
+        let msg = e.data,
-        if (e.data === '--init--') {
-            isinit = true;
-            dispatch.sync();
-            return;
-        }
-
-        let time_recv = time(),
-            msg = e.data || {},
             run = endpoints[msg.task],
+            done = false,
             send = {
-                data : function(data,direct) {
+                data: (data, direct) => {
-                    dispatch.send({
+                    if (canSend(data)) {
-                        seq: msg.seq,
+                        dispatch.send({ done: false, data: data }, direct);
-                        task: msg.task,
+                    }
-                        done: false,
-                        data: data
-                    }, direct);
                 },
-                done : function(data,direct) {
+                done: (data, direct) => {
-                    dispatch.send({
+                    if (canSend(data)) {
-                        seq: msg.seq,
+                        dispatch.send({ done: true, data: data }, direct);
-                        task: msg.task,
+                        done = true;
-                        done: true,
+                    }
-                        data: data
+                },
-                    }, direct);
+                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),
+                let data = run(msg.data, send);
-                    output = run(msg.data, send),
+                if (data) {
-                    time_send = time(),
+                    send.done(data);
-                    time_proc = time_send - time_recv;
+                }
-
+            } catch (error) {
-                if (output) dispatch.send({
+                console.trace(error.stack);
-                    seq: msg.seq,
+                dispatch.send({ data: {error: error.toString()}, done: true });
-                    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()});
             }
         } else {
-            console.log({worker_unhandled: e, msg, fn: dispatch[msg.task]});
+            console.log({worker_unhandled: e});
+            send.error(`no registered endpoint: ${msg.task}`);
         }
     }