30:T7f5,multi = { run; plain; // wait for single thread to finish return Promises.delay(wait).then(async ()=>{ let init = data; performance.mark('start'); let tasks = 2, threads = []; let pool = newchunked(init,tasks) // let size = Math.ceil(init.length / tasks); // let cuts = buffset(init,tasks); // calculat buffer offsets instead var src = `self.onmessage = function(e) { const t2 = performance.now(); const nums = new Float32Array(e.data.subarray); let sum = 0; for(let i = 0;i < nums.length; ++i) { sum = sum + nums[i]; } const subarray = new Float32Array([sum]).buffer;//results.buffer; const t3 = performance.now(); self.postMessage({tt:~~(t3-t2),subarray}, [subarray]); }`; let dat = [] var blob = new Blob([src], { type: "text/javascript",}); for (let n = 0; n < tasks; ++n) { var objectURL = URL.createObjectURL(blob); var thread = new Worker(objectURL); threads.push(thread); thread.postMessage({subarray:pool[n]},[pool[n]]) //let subbuf = init.slice(n * size, (n+1)*size).buffer; //let subbuf = init.buffer.slice(cuts[n].byteStart,cuts[n].byteEnd); //thread.postMessage({subarray:subbuf},[subbuf]) } // not using Promise.all here let queue = (Array.from({length:tasks},(_,i)=>new Promise((keep) => { threads[i].onmessage = function (e) { keep({tt:e.data.tt,subarray:new Float32Array(e.data.subarray)}) threads[i].terminate(); }; }))) invalidation.then(() => { threads.forEach(thread=>thread.terminate()); }) // initialise output array with n tasks when reducing, init.length when mapping let out = new Float32Array(tasks); let n = 0 for await (const q of queue) { out.set(q.subarray,n); n+= q.subarray.length; } // casting to array a little unnecessary here out = new Float32Array([out.reduce((a,b)=>a+b,0)/init.length]); performance.mark('stop') return { t:performance.measure('perf','start','stop').duration, r:out[0] } }) }31:T7f2,multiFloat = { // is faster because data is not transferred between threads run; plainFloat; // wait for single threaded to finish return Promises.delay(wait).then(async ()=>{ let t1 = performance.now(); let init = data; let tasks = 5, threads = []; let pool = newchunked(init,tasks) let func = d=>new Float32Array([d.map(d=>d).reduce((a,b)=>a+b,0)]); // optional function to insert into worker var src = `self.onmessage = function(e) { const t2 = performance.now(); // const fn = ${func.toString()}; // insert optional function //const nums = new Float32Array(e.data.subarray); const size = e.data.subarray let sum = 0; for(let i = 0; i < size ; ++i) { sum += Math.random(); } let subarray = new Float32Array(1); subarray[0] = sum; subarray = subarray.buffer; const t3 = performance.now(); self.postMessage({tt:~~(t3-t2),subarray}, [subarray]); }`; var blob = new Blob([src], { type: "text/javascript",}); for (let n = 0; n < tasks; ++n) { var objectURL = URL.createObjectURL(blob); var thread = new Worker(objectURL); threads.push(thread); thread.postMessage({subarray:pool[n].byteLength*0.25}/*, [pool[n]]*/) // transferring size instead of full array } let queue = Promise.all(Array.from({length:tasks},(_,i)=>new Promise((keep) => { threads[i].onmessage = function (e) { keep({tt:e.data.tt,subarray:new Float32Array(e.data.subarray)}) threads[i].terminate(); }; }))) invalidation.then(() => { threads.forEach(thread=>thread.terminate()); }) /*let out = new Float32Array(tasks); // tasks when reducing, init.length when mapping let n = 0 for await (const q of queue) { out.set(q.subarray,n); n+= q.subarray.length; } out = new Float32Array([out.reduce((a,b)=>a+b,0)/init.length])*/ let out = (await queue).map(d=>d.subarray[0]).reduce((a,b)=>a+b,0)/(init.length-1) return { t:~~(performance.now()-t1), r:out } }) }32:T50e,function newchunked(a = new Float32Array(), n = 5, balanced = true) { if (n < 2) return [a.buffer.slice(0)]; const len = a.length; const bytesPerElement = a.BYTES_PER_ELEMENT; const totalBytes = a.byteLength; const out = []; let i = 0; let size; function pushBuffer(start, end) { const byteStart = start * bytesPerElement const byteEnd = end * bytesPerElement const byteLength = byteEnd - byteStart; let t1 = performance.now() // formal copy /*const buffer = new ArrayBuffer(byteLength); const view = new Uint8Array(buffer); const sourceView = new Uint8Array(a.buffer, byteStart, byteLength); view.set(sourceView); out.push(buffer);*/ // easy copy out.push(a.buffer.slice(byteStart,byteEnd)); // console.log(~~(performance.now()-t1)) } if (len % n === 0) { size = Math.floor(len / n); while (i < len) { pushBuffer(i, i + size); i += size; } } else if (balanced) { while (i < len) { size = Math.ceil((len - i) / n--); pushBuffer(i, i + size); i += size; } } else { n--; size = Math.floor(len / n); if (len % size === 0) size--; while (i < size * n) { pushBuffer(i, i + size); i += size; } pushBuffer(i, len); } return out; }33:[] 2e:{"title":"Worker threads","cells":[{"id":0,"value":"# Worker threads\n## Prototyping 'multi-threaded' javascript","pinned":false,"hidden":false,"mode":"md"},{"id":5,"value":"___","pinned":false,"hidden":false,"mode":"md"},{"id":49,"value":"Under some circumstances, single threaded code can be 'parallelised' using worker threads. \nFor simple tasks such as summing, this *may* result in minor speed-ups depending on the device and environment (here using 5 workers).\nFor more complex tasks, larger speed-ups are expected, with arrays chunked in smaller pieces\nto be processed by each worker, taking intensive computations off the main thread.\nSpeed-ups noticeable around (2^25) items. \n\n> Artefact of `performance.now()` or true speedups?","pinned":false,"hidden":false,"mode":"md"},{"id":9,"value":"data = Float32Array.from(Array(~~(2**24.5)).values()).map((d,i)=>i)","pinned":false,"hidden":false,"mode":"ojs"},{"id":223,"value":"wait = 1000","pinned":false,"hidden":false,"mode":"ojs"},{"id":60,"value":"viewof run = Inputs.button('Rerun')","pinned":false,"hidden":false,"mode":"ojs"},{"id":46,"value":"___","pinned":false,"hidden":false,"mode":"md"},{"id":132,"value":"Summing and averaging integers (no guaranteed speed-up):","pinned":false,"hidden":false,"mode":"md"},{"id":12,"value":"plain = { run;\n\n return Promises.delay(wait).then(async ()=>{\n let t1 = performance.now();\n let init = data;\n \n let sum = 0;\n for (let i = 0; i < init.length; ++i) {\n sum += init[i]\n }\n \n sum = new Float32Array([sum/init.length])\n \n return {\n t:~~(performance.now()-t1),\n r:sum[0]\n }\n })\n}","pinned":false,"hidden":false,"mode":"ojs"},{"id":19,"value":"$30","pinned":false,"hidden":false,"mode":"ojs"},{"id":153,"value":"factorA = parseFloat((plain.t/multi.t).toPrecision(3))","pinned":false,"hidden":false,"mode":"ojs"},{"id":113,"value":"___","pinned":false,"hidden":false,"mode":"md"},{"id":140,"value":"Summing uncorrelated random floats and dividing by total elements:","pinned":false,"hidden":false,"mode":"md"},{"id":117,"value":"plainFloat = { run;multi;\n return Promises.delay(wait).then(async ()=>{\n let t1 = performance.now();\n let init = data;\n let sum = 0;\n \n for (let i = 0; i < init.length; ++i) {\n sum += Math.random()\n }\n \n sum = new Float32Array([sum/init.length])\n \n return {\n t:~~(performance.now()-t1),\n r:sum[0]\n }\n })\n \n}","pinned":false,"hidden":false,"mode":"ojs"},{"id":123,"value":"$31","pinned":false,"hidden":false,"mode":"ojs"},{"id":160,"value":"factorB = parseFloat((plainFloat.t/multiFloat.t).toPrecision(3))","pinned":false,"hidden":false,"mode":"ojs"},{"id":40,"value":"___\n\n### Util","pinned":false,"hidden":false,"mode":"md"},{"id":25,"value":"// adapted from: https://stackoverflow.com/a/8189268\nfunction chunked(a, n, balanced = true) {\n \n if (n < 2) return [a.slice(0).buffer];\n\n var len = a.length, \n out = [], i = 0,size;\n \n if (len % n === 0) {\n size = Math.floor(len / n);\n while (i < len) {\n out.push(a.slice(i, (i += size)).buffer);\n //out.push(a.subarray(i, (i += size)).buffer);\n\n }\n } else if (balanced) {\n while (i < len) {\n size = Math.ceil((len - i) / n--);\n out.push(a.slice(i, (i += size)).buffer);\n //out.push(a.subarray(i, (i += size)).buffer);\n }\n } else {\n n--;\n size = Math.floor(len / n);\n if (len % size === 0) size--;\n while (i < size * n) {\n out.push(a.slice(i, (i += size)).buffer);\n // out.push(a.subarray(i, (i += size)).buffer);\n }\n out.push(a.slice(size * n).buffer);\n // out.push(a.slice(size * n).buffer);\n }\n return out;\n}","pinned":false,"hidden":false,"mode":"ojs"},{"id":525,"value":"$32","pinned":true,"hidden":false,"mode":"ojs"},{"id":637,"value":"// reduce based chunker\n\nfunction copychunk(arr, n = 1) {\n n = Math.ceil(arr.length / n)\n if (n <= 0) throw new Error(\"Chunk size must be a positive integer\");\n return arr.reduce((acc, _, i) => {\n if (i % n === 0) acc.push(arr.slice(i, i + n).buffer);\n return acc;\n }, []);\n}","pinned":true,"hidden":false,"mode":"ojs"},{"id":716,"value":"// calculates buffer offsets given an array \n\nfunction buffset(a = new Float32Array(), n = 5, balanced = true) {\n if (n < 2) return [{byteStart:0,byteEnd:a.buffer.byteLength}];\n\n const len = a.length;\n const bytesPerElement = a.BYTES_PER_ELEMENT;\n const totalBytes = a.byteLength;\n const out = [];\n let i = 0;\n let size;\n\n function calcBuffer(start, end) {\n const byteStart = start * bytesPerElement\n const byteEnd = end * bytesPerElement\n const byteLength = byteEnd - byteStart;\n out.push({byteStart,byteEnd})\n }\n\n if (len % n === 0) {\n size = Math.floor(len / n);\n while (i < len) {\n calcBuffer(i, i + size);\n i += size;\n }\n } else if (balanced) {\n while (i < len) {\n size = Math.ceil((len - i) / n--);\n calcBuffer(i, i + size);\n i += size;\n }\n } else {\n n--;\n size = Math.floor(len / n);\n if (len % size === 0) size--;\n while (i < size * n) {\n calcBuffer(i, i + size);\n i += size;\n }\n calcBuffer(i, len);\n }\n\n return out;\n}","pinned":true,"hidden":false,"mode":"ojs"}],"files":"$W33","stdlib":"1","titling":"auto"}