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 start = System.nanoTime();
107 operationQueue.waitFor();
112 long duration = System.nanoTime()-start;
113 if(duration > 4000000000L) {
118 // long sss = System.nanoTime();
124 * Here we are actively processing updates from client.
125 * Maintain necessary caching here.
128 clusters.streamLRU.acquireMutex();
131 } catch (AcornAccessVerificationException | IllegalAcornStateException e) {
132 LOGGER.error("cluster chunk swapping failed", e);
134 clusters.streamLRU.releaseMutex();
136 clusters.csLRU.acquireMutex();
139 } catch (Throwable t) {
140 throw new IllegalAcornStateException(t);
142 clusters.csLRU.releaseMutex();
145 TimeLogger.log("Performed updates");
149 } catch (Throwable t) {
150 LOGGER.error("FATAL: MainProgram died unexpectedly", t);
152 deathBarrier.release();
157 static interface MainProgramRunnable {
158 void run() throws Exception;
159 default void error(Exception e) {
160 LOGGER.error("An error occured", e);
162 default void success() {}
165 private void runUpdates(TreeMap<ClusterUID, List<ClusterUpdateOperation>> updates) throws InterruptedException {
167 for(int i=0;i<CLUSTER_THREADS;i++)
168 updateSchedules[i].clear();
170 final Semaphore s = new Semaphore(0);
172 for(Map.Entry<ClusterUID, List<ClusterUpdateOperation>> entry : updates.entrySet()) {
173 ClusterUID key = entry.getKey();
174 int hash = key.hashCode() & (clusterUpdateThreads.length-1);
175 updateSchedules[hash].addAll(entry.getValue());
178 // final AtomicLong elapsed = new AtomicLong(0);
179 int acquireAmount = 0;
180 for(int i=0;i<CLUSTER_THREADS;i++) {
181 final List<ClusterUpdateOperation> ops = updateSchedules[i];
182 if (!ops.isEmpty()) {
184 clusterUpdateThreads[i].submit(new Callable<Object>() {
187 public Object call() throws Exception {
188 //long st = System.nanoTime();
190 for(ClusterUpdateOperation op : ops) {
198 // long duration = System.nanoTime()-st;
199 // elapsed.addAndGet(duration);
200 // double dur = 1e-9*duration;
202 // System.err.println("duration=" + dur + "s. " + ops.size());
208 s.acquire(acquireAmount);
213 * This shall be run when no updates are currently available.
215 private void runTasksIfEmpty() {
216 if(operationQueue.isEmpty()) {
217 List<MainProgramRunnable> todo = new ArrayList<>();
218 operationQueue.pumpTasks(todo);
219 for(MainProgramRunnable runnable : todo) {
223 } catch (Exception e) {
231 * This gets called when an idle period has been detected
233 private void checkIdle() throws IOException, IllegalAcornStateException, AcornAccessVerificationException {
235 // Was this a time-out or a new stream request?
236 if(operationQueue.isEmpty()) {
239 * We are idling here.
240 * Flush all caches gradually
243 // Write pending cs to disk
244 boolean written = clusters.csLRU.swapForced();
246 if(!operationQueue.isEmpty()) break;
247 written = clusters.csLRU.swapForced();
249 // Write pending chunks to disk
250 written = clusters.streamLRU.swapForced();
252 if(!operationQueue.isEmpty()) break;
253 written = clusters.streamLRU.swapForced();
255 // Write pending files to disk
256 written = clusters.fileLRU.swapForced();
258 if(!operationQueue.isEmpty()) break;
259 written = clusters.fileLRU.swapForced();
261 // Write pending clusters to disk
262 written = clusters.clusterLRU.swapForced();
264 if(!operationQueue.isEmpty()) break;
265 written = clusters.clusterLRU.swapForced();
268 client.tryMakeSnapshot();
276 * This schedules tasks to be run in MainProgram thread
277 * Called from other threads than MainProgram thread
280 void runIdle(MainProgramRunnable task) {
281 operationQueue.scheduleTask(task);
285 * Mutex for streamLRU is assumed here
288 private void swapChunks() throws AcornAccessVerificationException, IllegalAcornStateException {
290 // Cache chunks during update operations
291 while(clusters.streamLRU.swap(Long.MAX_VALUE, CHUNK_CACHE_SIZE));
294 private void swapCS() throws AcornAccessVerificationException, IllegalAcornStateException {
296 // Cache chunks during update operations
297 while(clusters.csLRU.swap(Long.MAX_VALUE, CHUNK_CACHE_SIZE));
301 * Called by DB client write threads
305 ClusterStreamChunk last = operationQueue.commitLast();
307 LOGGER.error("Trying to commit operation after MainProgram is closed! Operation is " + last);
313 * Called by DB client write threads
315 void schedule(ClusterUpdateOperation operation) throws IllegalAcornStateException {
318 LOGGER.error("Trying to schedule operation after MainProgram is closed! Operation is " + operation);
321 clusters.streamLRU.acquireMutex();
325 operationQueue.scheduleUpdate(operation);
328 } catch (IllegalAcornStateException e) {
330 } catch (Throwable t) {
331 throw new IllegalAcornStateException(t);
333 clusters.streamLRU.releaseMutex();
339 public void close() {
343 // This will wake up the sleeping beauty
344 operationQueue.scheduleTask(() -> {});
347 deathBarrier.acquire();
348 } catch (InterruptedException e) {
351 for (ExecutorService executor : clusterUpdateThreads)
354 for (int i = 0; i < clusterUpdateThreads.length; i++) {
356 ExecutorService executor = clusterUpdateThreads[i];
357 executor.awaitTermination(500, TimeUnit.MILLISECONDS);
358 clusterUpdateThreads[i] = null;
359 } catch (InterruptedException e) {
360 LOGGER.error("clusterUpdateThread[{}] termination interrupted", i, e);
365 void assertMainProgramThread() {
366 assert(Thread.currentThread().equals(mainProgramThread));
369 void assertNoMainProgramThread() {
370 assert(!Thread.currentThread().equals(mainProgramThread));