bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/Scheduler.java

   1 package org.simantics.db.impl.query;
   2
   3 import java.util.ArrayList;
   4 import java.util.HashMap;
   5 import java.util.Map;
   6 import java.util.concurrent.Semaphore;
   7
   8 class Scheduler extends ThreadLocal<TaskList> {
   9
  10     private final QueryProcessor processor;
  11
  12     private static final int MAX_TASKLIST_SIZE = 1<<8;
  13
  14     /*
  15      * New TaskList objects shall be assigned this size.
  16      * This shall grow exponentially upon task list creation to max size and shall reset to 1 when a write synchronizes the tasks.
  17      * The intention is to make sure that upon frequent syncs the tasks shall be scheduled
  18      */
  19     private int currentTaskListSize = 1;
  20
  21     /*
  22      * Anyone wishing to execute any tasks needs to hold this monitor
  23      */
  24     private Object taskExecutionMonitor = new Object();
  25
  26     /*
  27      * Anyone wishing to manipulate the list of dispatched tasks needs to hold this monitor
  28      */
  29     private Object taskDispatchMonitor = new Object();
  30
  31     /*
  32      * When a thread-specific TaskList becomes full it is added into this list for execution.
  33      * The executor thread takes TaskLists from this list.
  34      * The sync operation can also execute remaining tasks.
  35      */
  36     private ArrayList<TaskList> dispatchedTaskLists;
  37
  38     /*
  39      * This replicates the ThreadLocal map of TaskList objects to enable listing of tasks active queues
  40      */
  41     private final Map<Thread,TaskList> activeTaskLists = new HashMap<>();
  42
  43     /*
  44      * This semaphore is incremented each time s task list gets dispatched. The execution thread waits for the semaphore.
  45      */
  46     private final Semaphore taskDispatchNotifications = new Semaphore(0);
  47
  48     Scheduler(QueryProcessor processor) {
  49         this.processor = processor;
  50     }
  51
  52     /*
  53      * This gets called by multiple query threads
  54      */
  55     void accept(Runnable task) {
  56         TaskList l = get();
  57         l.add(task);
  58         if(l.shouldSchedule()) {
  59             remove();
  60             sendToExecution();
  61         }
  62     }
  63
  64     /*
  65      * A task list has reached full size and is moved into execution by the execution thread
  66      */
  67     void sendToExecution() {
  68         synchronized(activeTaskLists) {
  69             // This is dispatched so it should not anymore be in list of active TaskList objects.
  70             TaskList l = activeTaskLists.remove(Thread.currentThread());
  71             synchronized(taskDispatchMonitor) {
  72                 if(dispatchedTaskLists == null)
  73                     dispatchedTaskLists = new ArrayList<TaskList>();
  74                 dispatchedTaskLists.add(l);
  75             }
  76             if(currentTaskListSize < MAX_TASKLIST_SIZE)
  77                 currentTaskListSize = currentTaskListSize<<1;
  78         }
  79         // Release the execution thread
  80         taskDispatchNotifications.release();
  81     }
  82
  83     /*
  84      * Gets dispatched TaskList objects for execution.
  85      * Returns null when there's nothing to execute (and thread shall go to sleep).
  86      */
  87     ArrayList<TaskList> getDispatchedQueues() {
  88         synchronized(taskDispatchMonitor) {
  89             if(dispatchedTaskLists == null)
  90                 return null;
  91             ArrayList<TaskList> result = dispatchedTaskLists;
  92             dispatchedTaskLists = null;
  93             return result;
  94         }
  95     }
  96
  97     void start() {
  98
  99         Thread thread = new Thread() {
 100
 101             public void run() {
 102
 103                 // We execute tasks as long as the query processor is alive
 104                 while(!processor.isDisposed()) {
 105
 106                     try {
 107
 108                         // This is a mutex between this place and sync()
 109                         synchronized(taskExecutionMonitor) {
 110
 111                             // Execute busily while there is something to execute
 112                             while(true) {
 113
 114                                 ArrayList<TaskList> qs = getDispatchedQueues();
 115                                 if(qs == null) {
 116                                     // No tasks - go to sleep
 117                                     break;
 118                                 }
 119
 120                                 // Execute everything we got
 121                                 for(TaskList queue : qs) {
 122                                     for(Runnable r : queue) {
 123                                         r.run();
 124                                     }
 125                                 }
 126
 127                             }
 128
 129                         }
 130
 131                         // sendToExecution shall release this
 132                         taskDispatchNotifications.acquire();
 133
 134                     } catch (InterruptedException e) {
 135                         QueryListening.LOGGER.error("Unexpected interrupt", e);
 136                     }
 137
 138                 }
 139
 140             };
 141
 142         };
 143
 144         thread.setName("QueryListening");
 145         thread.start();
 146
 147     }
 148
 149     /*
 150      * ThreadLocal calls this to create a TaskList.
 151      * Called with first list per thread and also when accepting next task after sendToExecution.
 152      */
 153     @Override
 154     protected TaskList initialValue() {
 155         TaskList result = new TaskList(currentTaskListSize);
 156         synchronized(activeTaskLists) {
 157             activeTaskLists.put(Thread.currentThread(), result);
 158         }
 159         return result;
 160     }
 161
 162     /*
 163      * This makes sure that all tasks have been executed upon return.
 164      * It takes the taskExecutionMonitor from the thread and executes possible dispatched work and
 165      * also executes anything in active task lists. It resets the dispatch sizes of current and
 166      * subsequently created task lists to 1.
 167      */
 168     void sync() {
 169
 170         // This is a mutex between this place and execution thread
 171         synchronized(taskExecutionMonitor) {
 172
 173             currentTaskListSize = 1;
 174
 175             // Perform all dispatched work
 176             ArrayList<TaskList> qs = getDispatchedQueues();
 177             if(qs != null) {
 178                 for(TaskList queue : qs) {
 179                     for(Runnable r : queue) {
 180                         r.run();
 181                     }
 182                 }
 183             }
 184
 185             // Also perform any tasks in active queues.
 186             // This assumes that all query threads that send new tasks are now parked and shall not interfere.
 187             synchronized(activeTaskLists) {
 188                 for(TaskList queue : activeTaskLists.values()) {
 189                     // Run all tasks
 190                     for(Runnable r : queue) {
 191                         if(r != null)
 192                             r.run();
 193                     }
 194                     // Remove the tasks from list and reset the max size to 1.
 195                     queue.reset();
 196                 }
 197             }
 198
 199         }
 200
 201     }
 202
 203 }