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