1 package org.simantics.acorn;
3 import java.io.Closeable;
4 import java.io.IOException;
5 import java.util.ArrayList;
6 import java.util.Comparator;
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;
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;
26 public class MainProgram implements Runnable, Closeable {
28 private static final Logger LOGGER = LoggerFactory.getLogger(MainProgram.class);
30 private static final int CLUSTER_THREADS = 4;
31 private static final int CHUNK_CACHE_SIZE = 100;
33 private final GraphClientImpl2 client;
34 private final ExecutorService[] clusterUpdateThreads;
35 private final List<ClusterUpdateOperation>[] updateSchedules;
37 private Thread mainProgramThread;
39 private boolean alive = true;
40 private Semaphore deathBarrier = new Semaphore(0);
42 final ClusterManager clusters;
44 private final OperationQueue operationQueue = new OperationQueue(this);
46 static class ClusterThreadFactory implements ThreadFactory {
51 public ClusterThreadFactory(String name, boolean daemon) {
57 public Thread newThread(Runnable r) {
58 Thread thread = new Thread(r, name);
59 thread.setDaemon(daemon);
64 @SuppressWarnings("unchecked")
65 MainProgram(GraphClientImpl2 client, ClusterManager clusters) {
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>();
77 void startTransaction(long id) {
78 operationQueue.startTransaction(id);
81 private static Comparator<ClusterUID> clusterComparator = new Comparator<ClusterUID>() {
84 public int compare(ClusterUID o1, ClusterUID o2) {
85 return Long.compare(o1.second, o2.second);
92 mainProgramThread = Thread.currentThread();
99 TreeMap<ClusterUID, List<ClusterUpdateOperation>> updates = new TreeMap<ClusterUID, List<ClusterUpdateOperation>>(clusterComparator);
101 operationQueue.pumpUpdates(updates);
103 if(updates.isEmpty()) {
105 long duration = operationQueue.waitFor();
110 if(duration > 4000000000L) {
116 // long sss = System.nanoTime();
122 * Here we are actively processing updates from client.
123 * Maintain necessary caching here.
126 clusters.streamLRU.acquireMutex();
129 } catch (AcornAccessVerificationException | IllegalAcornStateException e) {
130 LOGGER.error("cluster chunk swapping failed", e);
132 clusters.streamLRU.releaseMutex();
134 clusters.csLRU.acquireMutex();
137 } catch (Throwable t) {
138 throw new IllegalAcornStateException(t);
140 clusters.csLRU.releaseMutex();
143 TimeLogger.log("Performed updates");
147 } catch (Throwable t) {
148 LOGGER.error("FATAL: MainProgram died unexpectedly", t);
150 deathBarrier.release();
155 static interface MainProgramRunnable {
156 void run() throws Exception;
157 default void error(Exception e) {
158 LOGGER.error("An error occured", e);
160 default void success() {}
163 private void runUpdates(TreeMap<ClusterUID, List<ClusterUpdateOperation>> updates) throws InterruptedException {
165 for(int i=0;i<CLUSTER_THREADS;i++)
166 updateSchedules[i].clear();
168 final Semaphore s = new Semaphore(0);
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());
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()) {
182 clusterUpdateThreads[i].submit(new Callable<Object>() {
185 public Object call() throws Exception {
186 //long st = System.nanoTime();
188 for(ClusterUpdateOperation op : ops) {
196 // long duration = System.nanoTime()-st;
197 // elapsed.addAndGet(duration);
198 // double dur = 1e-9*duration;
200 // System.err.println("duration=" + dur + "s. " + ops.size());
206 s.acquire(acquireAmount);
211 * This shall be run when no updates are currently available.
213 private void runTasksIfEmpty() {
214 if(operationQueue.isEmpty()) {
215 List<MainProgramRunnable> todo = new ArrayList<>();
216 operationQueue.pumpTasks(todo);
217 for(MainProgramRunnable runnable : todo) {
221 } catch (Exception e) {
229 * This gets called when an idle period has been detected
231 private void checkIdle() throws IOException, IllegalAcornStateException, AcornAccessVerificationException {
233 // Was this a time-out or a new stream request?
234 if(operationQueue.isEmpty()) {
237 * We are idling here.
238 * Flush all caches gradually
241 // Write pending cs to disk
242 boolean written = clusters.csLRU.swapForced();
244 if(!operationQueue.isEmpty()) break;
245 written = clusters.csLRU.swapForced();
247 // Write pending chunks to disk
248 written = clusters.streamLRU.swapForced();
250 if(!operationQueue.isEmpty()) break;
251 written = clusters.streamLRU.swapForced();
253 // Write pending files to disk
254 written = clusters.fileLRU.swapForced();
256 if(!operationQueue.isEmpty()) break;
257 written = clusters.fileLRU.swapForced();
259 // Write pending clusters to disk
260 written = clusters.clusterLRU.swapForced();
262 if(!operationQueue.isEmpty()) break;
263 written = clusters.clusterLRU.swapForced();
266 client.tryMakeSnapshot();
274 * This schedules tasks to be run in MainProgram thread
275 * Called from other threads than MainProgram thread
278 void runIdle(MainProgramRunnable task) {
279 operationQueue.scheduleTask(task);
283 * Mutex for streamLRU is assumed here
286 private void swapChunks() throws AcornAccessVerificationException, IllegalAcornStateException {
288 // Cache chunks during update operations
289 while(clusters.streamLRU.swap(Long.MAX_VALUE, CHUNK_CACHE_SIZE));
292 private void swapCS() throws AcornAccessVerificationException, IllegalAcornStateException {
294 // Cache chunks during update operations
295 while(clusters.csLRU.swap(Long.MAX_VALUE, CHUNK_CACHE_SIZE));
299 * Called by DB client write threads
303 ClusterStreamChunk last = operationQueue.commitLast();
305 LOGGER.error("Trying to commit operation after MainProgram is closed! Operation is " + last);
311 * Called by DB client write threads
313 void schedule(ClusterUpdateOperation operation) throws IllegalAcornStateException {
316 LOGGER.error("Trying to schedule operation after MainProgram is closed! Operation is " + operation);
319 clusters.streamLRU.acquireMutex();
323 operationQueue.scheduleUpdate(operation);
326 } catch (IllegalAcornStateException e) {
328 } catch (Throwable t) {
329 throw new IllegalAcornStateException(t);
331 clusters.streamLRU.releaseMutex();
337 public void close() {
341 // This will wake up the sleeping beauty
342 operationQueue.scheduleTask(() -> {});
345 deathBarrier.acquire();
346 } catch (InterruptedException e) {
349 for (ExecutorService executor : clusterUpdateThreads)
352 for (int i = 0; i < clusterUpdateThreads.length; i++) {
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);
363 void assertMainProgramThread() {
364 assert(Thread.currentThread().equals(mainProgramThread));
367 void assertNoMainProgramThread() {
368 assert(!Thread.currentThread().equals(mainProgramThread));