ext/wasm/tests/opfs/concurrency/worker.js

   1 importScripts(
   2   (new URL(self.location.href).searchParams).get('sqlite3.dir') + '/sqlite3.js'
   3 );
   4 self.sqlite3InitModule().then(async function(sqlite3){
   5   const urlArgs = new URL(self.location.href).searchParams;
   6   const options = {
   7     workerName: urlArgs.get('workerId') || Math.round(Math.random()*10000),
   8     unlockAsap: urlArgs.get('opfs-unlock-asap') || 0 /*EXPERIMENTAL*/
   9   };
  10   const wPost = (type,...payload)=>{
  11     postMessage({type, worker: options.workerName, payload});
  12   };
  13   const stdout = (...args)=>wPost('stdout',...args);
  14   const stderr = (...args)=>wPost('stderr',...args);
  15   if(!sqlite3.opfs){
  16     stderr("OPFS support not detected. Aborting.");
  17     return;
  18   }
  19
  20   const wait = async (ms)=>{
  21     return new Promise((resolve)=>setTimeout(resolve,ms));
  22   };
  23
  24   const dbName = 'concurrency-tester.db';
  25   if(urlArgs.has('unlink-db')){
  26     await sqlite3.opfs.unlink(dbName);
  27     stdout("Unlinked",dbName);
  28   }
  29   wPost('loaded');
  30   let db;
  31   const interval = Object.assign(Object.create(null),{
  32     delay: urlArgs.has('interval') ? (+urlArgs.get('interval') || 750) : 750,
  33     handle: undefined,
  34     count: 0
  35   });
  36   const finish = ()=>{
  37     if(db){
  38       if(!db.pointer) return;
  39       db.close();
  40     }
  41     if(interval.error){
  42       wPost('failed',"Ending work after interval #"+interval.count,
  43             "due to error:",interval.error);
  44     }else{
  45       wPost('finished',"Ending work after",interval.count,"intervals.");
  46     }
  47   };
  48   const run = async function(){
  49     db = new sqlite3.oo1.OpfsDb({
  50       filename: 'file:'+dbName+'?opfs-unlock-asap='+options.unlockAsap,
  51       flags: 'c'
  52     });
  53     sqlite3.capi.sqlite3_busy_timeout(db.pointer, 5000);
  54     db.transaction((db)=>{
  55       db.exec([
  56         "create table if not exists t1(w TEXT UNIQUE ON CONFLICT REPLACE,v);",
  57         "create table if not exists t2(w TEXT UNIQUE ON CONFLICT REPLACE,v);"
  58       ]);
  59     });
  60
  61     const maxIterations =
  62           urlArgs.has('iterations') ? (+urlArgs.get('iterations') || 10) : 10;
  63     stdout("Starting interval-based db updates with delay of",interval.delay,"ms.");
  64     const doWork = async ()=>{
  65       const tm = new Date().getTime();
  66       ++interval.count;
  67       const prefix = "v(#"+interval.count+")";
  68       stdout("Setting",prefix,"=",tm);
  69       try{
  70         db.exec({
  71           sql:"INSERT OR REPLACE INTO t1(w,v) VALUES(?,?)",
  72           bind: [options.workerName, new Date().getTime()]
  73         });
  74         //stdout("Set",prefix);
  75       }catch(e){
  76         interval.error = e;
  77       }
  78     };
  79     if(1){/*use setInterval()*/
  80       setTimeout(async function timer(){
  81         await doWork();
  82         if(interval.error || maxIterations === interval.count){
  83           finish();
  84         }else{
  85           setTimeout(timer, interval.delay);
  86         }
  87       }, interval.delay);
  88     }else{
  89       /*This approach provides no concurrency whatsoever: each worker
  90         is run to completion before any others can work.*/
  91       let i;
  92       for(i = 0; i < maxIterations; ++i){
  93         await doWork();
  94         if(interval.error) break;
  95         await wait(interval.ms);
  96       }
  97       finish();
  98     }
  99   }/*run()*/;
 100
 101   self.onmessage = function({data}){
 102     switch(data.type){
 103         case 'run': run().catch((e)=>{
 104           if(!interval.error) interval.error = e;
 105           finish();
 106         });
 107           break;
 108         default:
 109           stderr("Unhandled message type '"+data.type+"'.");
 110           break;
 111     }
 112   };
 113 });