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<>();
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();
96 TreeMap<ClusterUID, List<ClusterUpdateOperation>> updates = new TreeMap<>(clusterComparator);
101 if (!updates.isEmpty())
103 operationQueue.pumpUpdates(updates);
105 if(updates.isEmpty()) {
107 long duration = operationQueue.waitFor();
112 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 if (updates.isEmpty())
173 final Semaphore s = new Semaphore(0);
175 for(Map.Entry<ClusterUID, List<ClusterUpdateOperation>> entry : updates.entrySet()) {
176 ClusterUID key = entry.getKey();
177 int hash = key.hashCode() & (clusterUpdateThreads.length-1);
178 updateSchedules[hash].addAll(entry.getValue());
181 // final AtomicLong elapsed = new AtomicLong(0);
182 int acquireAmount = 0;
183 for(int i=0;i<CLUSTER_THREADS;i++) {
184 final List<ClusterUpdateOperation> ops = updateSchedules[i];
185 if (!ops.isEmpty()) {
187 clusterUpdateThreads[i].submit(new Callable<Object>() {
190 public Object call() throws Exception {
191 //long st = System.nanoTime();
193 for(ClusterUpdateOperation op : ops) {
201 // long duration = System.nanoTime()-st;
202 // elapsed.addAndGet(duration);
203 // double dur = 1e-9*duration;
205 // System.err.println("duration=" + dur + "s. " + ops.size());
211 s.acquire(acquireAmount);
216 * This shall be run when no updates are currently available.
218 private void runTasksIfEmpty() {
219 if(operationQueue.isEmpty()) {
220 List<MainProgramRunnable> todo = new ArrayList<>();
221 operationQueue.pumpTasks(todo);
222 for(MainProgramRunnable runnable : todo) {
226 } catch (Exception e) {
234 * This gets called when an idle period has been detected
236 private void checkIdle() throws IOException, IllegalAcornStateException, AcornAccessVerificationException {
238 // Was this a time-out or a new stream request?
239 if(operationQueue.isEmpty()) {
242 * We are idling here.
243 * Flush all caches gradually
246 // Write pending cs to disk
247 boolean written = clusters.csLRU.swapForced();
249 if(!operationQueue.isEmpty()) break;
250 written = clusters.csLRU.swapForced();
252 // Write pending chunks to disk
253 written = clusters.streamLRU.swapForced();
255 if(!operationQueue.isEmpty()) break;
256 written = clusters.streamLRU.swapForced();
258 // Write pending files to disk
259 written = clusters.fileLRU.swapForced();
261 if(!operationQueue.isEmpty()) break;
262 written = clusters.fileLRU.swapForced();
264 // Write pending clusters to disk
265 written = clusters.clusterLRU.swapForced();
267 if(!operationQueue.isEmpty()) break;
268 written = clusters.clusterLRU.swapForced();
271 client.tryMakeSnapshot();
279 * This schedules tasks to be run in MainProgram thread
280 * Called from other threads than MainProgram thread
283 void runIdle(MainProgramRunnable task) {
284 operationQueue.scheduleTask(task);
288 * Mutex for streamLRU is assumed here
291 private void swapChunks() throws AcornAccessVerificationException, IllegalAcornStateException {
293 // Cache chunks during update operations
294 while(clusters.streamLRU.swap(Long.MAX_VALUE, CHUNK_CACHE_SIZE));
297 private void swapCS() throws AcornAccessVerificationException, IllegalAcornStateException {
299 // Cache chunks during update operations
300 while(clusters.csLRU.swap(Long.MAX_VALUE, CHUNK_CACHE_SIZE));
304 * Called by DB client write threads
308 ClusterStreamChunk last = operationQueue.commitLast();
310 LOGGER.error("Trying to commit operation after MainProgram is closed! Operation is " + last);
316 * Called by DB client write threads
318 void schedule(ClusterUpdateOperation operation) throws IllegalAcornStateException {
321 LOGGER.error("Trying to schedule operation after MainProgram is closed! Operation is " + operation);
324 clusters.streamLRU.acquireMutex();
328 operationQueue.scheduleUpdate(operation);
331 } catch (IllegalAcornStateException e) {
333 } catch (Throwable t) {
334 throw new IllegalAcornStateException(t);
336 clusters.streamLRU.releaseMutex();
342 public void close() {
346 // This will wake up the sleeping beauty
347 operationQueue.scheduleTask(() -> {});
350 deathBarrier.acquire();
351 } catch (InterruptedException e) {
354 for (ExecutorService executor : clusterUpdateThreads)
357 for (int i = 0; i < clusterUpdateThreads.length; i++) {
359 ExecutorService executor = clusterUpdateThreads[i];
360 executor.awaitTermination(500, TimeUnit.MILLISECONDS);
361 clusterUpdateThreads[i] = null;
362 } catch (InterruptedException e) {
363 LOGGER.error("clusterUpdateThread[{}] termination interrupted", i, e);
368 void assertMainProgramThread() {
369 assert(Thread.currentThread().equals(mainProgramThread));
372 void assertNoMainProgramThread() {
373 assert(!Thread.currentThread().equals(mainProgramThread));