bundles/org.simantics.acorn/src/org/simantics/acorn/MainProgram.java

   1 package org.simantics.acorn;
   2
   3 import java.io.Closeable;
   4 import java.util.ArrayList;
   5 import java.util.Comparator;
   6 import java.util.LinkedList;
   7 import java.util.List;
   8 import java.util.Map;
   9 import java.util.TreeMap;
  10 import java.util.concurrent.Callable;
  11 import java.util.concurrent.ExecutorService;
  12 import java.util.concurrent.Executors;
  13 import java.util.concurrent.Semaphore;
  14 import java.util.concurrent.ThreadFactory;
  15 import java.util.concurrent.TimeUnit;
  16
  17 import org.simantics.acorn.exception.AcornAccessVerificationException;
  18 import org.simantics.acorn.exception.IllegalAcornStateException;
  19 import org.simantics.acorn.lru.ClusterStreamChunk;
  20 import org.simantics.acorn.lru.ClusterUpdateOperation;
  21 import org.simantics.db.service.ClusterUID;
  22 import org.simantics.utils.logging.TimeLogger;
  23
  24 public class MainProgram implements Runnable, Closeable {
  25
  26         private static final int CLUSTER_THREADS = 4;
  27         private static final int CHUNK_CACHE_SIZE = 100;
  28
  29         private final GraphClientImpl2 client;
  30         private final ClusterManager clusters;
  31         private final ExecutorService[] clusterUpdateThreads;
  32     private final List<ClusterUpdateOperation>[] updateSchedules;
  33
  34         private int residentOperationBytes = 0;
  35         private long currentChangeSetId = -1;
  36         private int nextChunkId = 0;
  37         private boolean alive = true;
  38         private Semaphore deathBarrier = new Semaphore(0);
  39
  40         final Semaphore mutex = new Semaphore(1);
  41         final LinkedList<ClusterStreamChunk> operations = new LinkedList<>();
  42
  43         static class ClusterThreadFactory implements ThreadFactory {
  44
  45                 final String name;
  46                 final boolean daemon;
  47
  48                 public ClusterThreadFactory(String name, boolean daemon) {
  49                         this.name = name;
  50                         this.daemon = daemon;
  51                 }
  52
  53                 @Override
  54                 public Thread newThread(Runnable r) {
  55                         Thread thread = new Thread(r, name);
  56                         thread.setDaemon(daemon);
  57                         return thread;
  58                 }
  59         }
  60
  61         public MainProgram(GraphClientImpl2 client, ClusterManager clusters) {
  62
  63                 this.client = client;
  64                 this.clusters = clusters;
  65                 this.clusterUpdateThreads = new ExecutorService[CLUSTER_THREADS];
  66                 this.updateSchedules = new ArrayList[CLUSTER_THREADS];
  67                 for(int i=0;i<clusterUpdateThreads.length;i++) {
  68                         clusterUpdateThreads[i] = Executors.newSingleThreadExecutor(new ClusterThreadFactory("Cluster Updater " + (i+1), false));
  69                         updateSchedules[i] = new ArrayList<ClusterUpdateOperation>();
  70                 }
  71         }
  72
  73         public void startTransaction(long id) {
  74                 currentChangeSetId = id;
  75                 nextChunkId = 0;
  76         }
  77
  78         private static Comparator<ClusterUID> clusterComparator = new Comparator<ClusterUID>() {
  79
  80                 @Override
  81                 public int compare(ClusterUID o1, ClusterUID o2) {
  82                         return Long.compare(o1.second, o2.second);
  83                 }
  84         };
  85
  86         @Override
  87         public void run() {
  88                 try {
  89
  90                         mutex.acquire();
  91                         main:
  92                         while(alive) {
  93
  94                                 TreeMap<ClusterUID, List<ClusterUpdateOperation>> updates = new TreeMap<ClusterUID, List<ClusterUpdateOperation>>(clusterComparator);
  95
  96                                 synchronized(MainProgram.this) {
  97
  98                                         while(!operations.isEmpty() && updates.size() < 100) {
  99
 100                                                 ClusterStreamChunk chunk = operations.pollFirst();
 101
 102                                                 for(int i=chunk.nextToProcess;i<chunk.operations.size();i++) {
 103                                                         ClusterUpdateOperation o = chunk.operations.get(i);
 104                                                         ClusterUID uid = o.uid;
 105                                                         List<ClusterUpdateOperation> ops = updates.get(uid);
 106                                                         if(ops == null) {
 107                                                                 ops = new ArrayList<ClusterUpdateOperation>();
 108                                                                 updates.put(uid, ops);
 109                                                         }
 110                                                         ops.add(o);
 111                                                 }
 112
 113                                                 chunk.nextToProcess = chunk.operations.size();
 114
 115                                                 if(!chunk.isCommitted()) {
 116                                                         assert(operations.isEmpty());
 117                                                         operations.add(chunk);
 118                                                         break;
 119                                                 }
 120
 121                                         }
 122
 123                                         if(updates.isEmpty()) {
 124                                                 try {
 125                                                         long start = System.nanoTime();
 126                                                         mutex.release();
 127                                                         MainProgram.this.wait(5000);
 128                                                         mutex.acquire();
 129                                                         if (!alive)
 130                                                                 break main;
 131                                                         long duration = System.nanoTime()-start;
 132                                                         if(duration > 4000000000L) {
 133
 134                                                                 // Was this a time-out or a new stream request?
 135                                                                 if(operations.isEmpty()) {
 136
 137                                                                         /*
 138                                                                          * We are idling here.
 139                                                                          * Flush all caches gradually
 140                                                                          */
 141
 142                                                                         // Write pending cs to disk
 143                                                                         boolean written = clusters.csLRU.swapForced();
 144                                                                         while(written) {
 145                                                                                 if(!updates.isEmpty()) break;
 146                                                                                 written = clusters.csLRU.swapForced();
 147                                                                         }
 148                                                                         // Write pending chunks to disk
 149                                                                         written = clusters.streamLRU.swapForced();
 150                                                                         while(written) {
 151                                                                                 if(!updates.isEmpty()) break;
 152                                                                                 written = clusters.streamLRU.swapForced();
 153                                                                         }
 154                                                                         // Write pending files to disk
 155                                                                         written = clusters.fileLRU.swapForced();
 156                                                                         while(written) {
 157                                                                                 if(!updates.isEmpty()) break;
 158                                                                                 written = clusters.fileLRU.swapForced();
 159                                                                         }
 160                                                                         // Write pending clusters to disk
 161                                                                         written = clusters.clusterLRU.swapForced();
 162                                                                         while(written) {
 163                                                                                 if(!updates.isEmpty()) break;
 164                                                                                 written = clusters.clusterLRU.swapForced();
 165                                                                         }
 166
 167                                                                         client.tryMakeSnapshot();
 168                                                                 }
 169                                                         }
 170                                                 } catch (InterruptedException e) {
 171                                                         e.printStackTrace();
 172                                                 }
 173                                         }
 174                                 }
 175
 176 //                              long sss = System.nanoTime();
 177
 178                                 for(int i=0;i<CLUSTER_THREADS;i++)
 179                                         updateSchedules[i].clear();
 180
 181                                 final Semaphore s = new Semaphore(0);
 182
 183                                 for(Map.Entry<ClusterUID, List<ClusterUpdateOperation>> entry : updates.entrySet()) {
 184                                         ClusterUID key = entry.getKey();
 185                                         int hash = key.hashCode() & (clusterUpdateThreads.length-1);
 186                                         updateSchedules[hash].addAll(entry.getValue());
 187                                 }
 188
 189                                 //                              final AtomicLong elapsed = new AtomicLong(0);
 190                                 int acquireAmount = 0;
 191                                 for(int i=0;i<CLUSTER_THREADS;i++) {
 192                                         final List<ClusterUpdateOperation> ops = updateSchedules[i];
 193                                         if (!ops.isEmpty()) {
 194                                                 acquireAmount++;
 195                                                 clusterUpdateThreads[i].submit(new Callable<Object>() {
 196
 197                             @Override
 198                             public Object call() throws Exception {
 199                                 //long st = System.nanoTime();
 200                                 try {
 201                                     for(ClusterUpdateOperation op : ops) {
 202                                         op.run();
 203                                     }
 204                                 } finally {
 205                                     s.release();
 206                                 }
 207                                 return null;
 208
 209     //                          long duration = System.nanoTime()-st;
 210     //                          elapsed.addAndGet(duration);
 211     //                          double dur = 1e-9*duration;
 212     //                          if(dur > 0.05)
 213     //                              System.err.println("duration=" + dur + "s. " + ops.size());
 214                             }
 215                         });
 216                                         }
 217                                 }
 218
 219                                 s.acquire(acquireAmount);
 220
 221                                 /*
 222                                  * Here we are actively processing updates from client.
 223                                  * Maintain necessary caching here.
 224                                  */
 225
 226                                 clusters.streamLRU.acquireMutex();
 227                                 try {
 228                                         swapChunks();
 229                                 } catch (AcornAccessVerificationException | IllegalAcornStateException e) {
 230                                     e.printStackTrace();
 231                                 } finally {
 232                                         clusters.streamLRU.releaseMutex();
 233                                 }
 234                                 clusters.csLRU.acquireMutex();
 235                                 try {
 236                                         swapCS();
 237                                 } catch (Throwable t) {
 238                                         throw new IllegalAcornStateException(t);
 239                                 } finally {
 240                                         clusters.csLRU.releaseMutex();
 241                                 }
 242
 243                                 TimeLogger.log("Performed updates");
 244
 245                         }
 246
 247                 } catch (Throwable t) {
 248                         t.printStackTrace();
 249                 } finally {
 250                         deathBarrier.release();
 251                 }
 252         }
 253
 254         static interface MainProgramRunnable {
 255
 256                 public void run() throws Exception;
 257                 public void done();
 258
 259         }
 260
 261         public Exception runIdle(MainProgramRunnable runnable) {
 262                 try {
 263                         mutex.acquire();
 264                         runnable.run();
 265                         return null;
 266                 } catch (Exception e) {
 267                         return e;
 268                 } finally {
 269                         runnable.done();
 270                         mutex.release();
 271                 }
 272         }
 273
 274         /*
 275          * Mutex for streamLRU is assumed here
 276          *
 277          */
 278         private void swapChunks() throws AcornAccessVerificationException, IllegalAcornStateException {
 279
 280                 // Cache chunks during update operations
 281                 boolean written = clusters.streamLRU.swap(Integer.MAX_VALUE, CHUNK_CACHE_SIZE);
 282                 while(written) {
 283                         written = clusters.streamLRU.swap(Integer.MAX_VALUE, CHUNK_CACHE_SIZE);
 284                 }
 285         }
 286
 287         private void swapCS() throws AcornAccessVerificationException, IllegalAcornStateException {
 288
 289                 // Cache chunks during update operations
 290                 boolean written = clusters.csLRU.swap(Integer.MAX_VALUE, CHUNK_CACHE_SIZE);
 291                 while(written) {
 292                         written = clusters.csLRU.swap(Integer.MAX_VALUE, CHUNK_CACHE_SIZE);
 293                 }
 294         }
 295
 296         public synchronized void committed() {
 297
 298                 ClusterStreamChunk last = operations.isEmpty() ? null : operations.getLast();
 299         if (!alive) {
 300             System.err.println("Trying to commit operation after MainProgram is closed! Operation is " + last);
 301 //          return;
 302         }
 303                 if(last != null) last.commit();
 304
 305         }
 306
 307         public synchronized void schedule(ClusterUpdateOperation operation) throws IllegalAcornStateException {
 308             if (!alive) {
 309                 System.err.println("Trying to schedule operation after MainProgram is closed! Operation is " + operation);
 310 //              return;
 311             }
 312                 clusters.streamLRU.acquireMutex();
 313
 314                 try {
 315
 316                         ClusterStreamChunk last = operations.isEmpty() ? null : operations.getLast();
 317                         if(last == null || last.isCommitted()) {
 318                                 String id = "" + currentChangeSetId + "-" + nextChunkId++;
 319                                 last = new ClusterStreamChunk(clusters, clusters.streamLRU, id);
 320                                 operations.add(last);
 321                         }
 322
 323                         String chunkId = last.getKey();
 324                         int chunkOffset = last.operations.size();
 325                         operation.scheduled(chunkId + "." + chunkOffset);
 326
 327                         last.addOperation(operation);
 328
 329                         swapChunks();
 330
 331                         notifyAll();
 332                 } catch (IllegalAcornStateException e) {
 333                     throw e;
 334                 } catch (Throwable t) {
 335                         throw new IllegalAcornStateException(t);
 336                 } finally {
 337                         clusters.streamLRU.releaseMutex();
 338                 }
 339         }
 340
 341     @Override
 342     public void close() {
 343         alive = false;
 344         synchronized (this) {
 345             notifyAll();
 346         }
 347         try {
 348             deathBarrier.acquire();
 349         } catch (InterruptedException e) {
 350         }
 351
 352         for (ExecutorService executor : clusterUpdateThreads)
 353             executor.shutdown();
 354
 355         for (int i = 0; i < clusterUpdateThreads.length; i++) {
 356             try {
 357                 ExecutorService executor  = clusterUpdateThreads[i];
 358                 executor.awaitTermination(500, TimeUnit.MILLISECONDS);
 359                 clusterUpdateThreads[i] = null;
 360             } catch (InterruptedException e) {
 361                 e.printStackTrace();
 362             }
 363         }
 364     }
 365
 366 }