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 MainProgram(GraphClientImpl2 client, ClusterManager clusters) {
67 this.clusters = clusters;
68 this.clusterUpdateThreads = new ExecutorService[CLUSTER_THREADS];
69 this.updateSchedules = new ArrayList[CLUSTER_THREADS];
70 for(int i=0;i<clusterUpdateThreads.length;i++) {
71 clusterUpdateThreads[i] = Executors.newSingleThreadExecutor(new ClusterThreadFactory("Cluster Updater " + (i+1), false));
72 updateSchedules[i] = new ArrayList<ClusterUpdateOperation>();
76 void startTransaction(long id) {
77 operationQueue.startTransaction(id);
80 private static Comparator<ClusterUID> clusterComparator = new Comparator<ClusterUID>() {
83 public int compare(ClusterUID o1, ClusterUID o2) {
84 return Long.compare(o1.second, o2.second);
91 mainProgramThread = Thread.currentThread();
98 TreeMap<ClusterUID, List<ClusterUpdateOperation>> updates = new TreeMap<ClusterUID, List<ClusterUpdateOperation>>(clusterComparator);
100 operationQueue.pumpUpdates(updates);
102 if(updates.isEmpty()) {
104 long start = System.nanoTime();
106 operationQueue.waitFor();
111 long duration = System.nanoTime()-start;
112 if(duration > 4000000000L) {
117 // long sss = System.nanoTime();
123 * Here we are actively processing updates from client.
124 * Maintain necessary caching here.
127 clusters.streamLRU.acquireMutex();
130 } catch (AcornAccessVerificationException | IllegalAcornStateException e) {
133 clusters.streamLRU.releaseMutex();
135 clusters.csLRU.acquireMutex();
138 } catch (Throwable t) {
139 throw new IllegalAcornStateException(t);
141 clusters.csLRU.releaseMutex();
144 TimeLogger.log("Performed updates");
148 } catch (Throwable t) {
151 deathBarrier.release();
156 static interface MainProgramRunnable {
157 void run() throws Exception;
158 default void error(Exception e) {
159 LOGGER.error("An error occured", e);
161 default void success() {}
164 private void runUpdates(TreeMap<ClusterUID, List<ClusterUpdateOperation>> updates) throws InterruptedException {
166 for(int i=0;i<CLUSTER_THREADS;i++)
167 updateSchedules[i].clear();
169 final Semaphore s = new Semaphore(0);
171 for(Map.Entry<ClusterUID, List<ClusterUpdateOperation>> entry : updates.entrySet()) {
172 ClusterUID key = entry.getKey();
173 int hash = key.hashCode() & (clusterUpdateThreads.length-1);
174 updateSchedules[hash].addAll(entry.getValue());
177 // final AtomicLong elapsed = new AtomicLong(0);
178 int acquireAmount = 0;
179 for(int i=0;i<CLUSTER_THREADS;i++) {
180 final List<ClusterUpdateOperation> ops = updateSchedules[i];
181 if (!ops.isEmpty()) {
183 clusterUpdateThreads[i].submit(new Callable<Object>() {
186 public Object call() throws Exception {
187 //long st = System.nanoTime();
189 for(ClusterUpdateOperation op : ops) {
197 // long duration = System.nanoTime()-st;
198 // elapsed.addAndGet(duration);
199 // double dur = 1e-9*duration;
201 // System.err.println("duration=" + dur + "s. " + ops.size());
207 s.acquire(acquireAmount);
212 * This shall be run when no updates are currently available.
214 private void runTasksIfEmpty() {
215 if(operationQueue.isEmpty()) {
216 List<MainProgramRunnable> todo = new ArrayList<>();
217 operationQueue.pumpTasks(todo);
218 for(MainProgramRunnable runnable : todo) {
222 } catch (Exception e) {
230 * This gets called when an idle period has been detected
232 private void checkIdle() throws IOException, IllegalAcornStateException, AcornAccessVerificationException {
234 // Was this a time-out or a new stream request?
235 if(operationQueue.isEmpty()) {
238 * We are idling here.
239 * Flush all caches gradually
242 // Write pending cs to disk
243 boolean written = clusters.csLRU.swapForced();
245 if(!operationQueue.isEmpty()) break;
246 written = clusters.csLRU.swapForced();
248 // Write pending chunks to disk
249 written = clusters.streamLRU.swapForced();
251 if(!operationQueue.isEmpty()) break;
252 written = clusters.streamLRU.swapForced();
254 // Write pending files to disk
255 written = clusters.fileLRU.swapForced();
257 if(!operationQueue.isEmpty()) break;
258 written = clusters.fileLRU.swapForced();
260 // Write pending clusters to disk
261 written = clusters.clusterLRU.swapForced();
263 if(!operationQueue.isEmpty()) break;
264 written = clusters.clusterLRU.swapForced();
267 client.tryMakeSnapshot();
275 * This schedules tasks to be run in MainProgram thread
276 * Called from other threads than MainProgram thread
279 void runIdle(MainProgramRunnable task) {
280 operationQueue.scheduleTask(task);
284 * Mutex for streamLRU is assumed here
287 private void swapChunks() throws AcornAccessVerificationException, IllegalAcornStateException {
289 // Cache chunks during update operations
290 while(clusters.streamLRU.swap(Long.MAX_VALUE, CHUNK_CACHE_SIZE));
293 private void swapCS() throws AcornAccessVerificationException, IllegalAcornStateException {
295 // Cache chunks during update operations
296 while(clusters.csLRU.swap(Long.MAX_VALUE, CHUNK_CACHE_SIZE));
300 * Called by DB client write threads
304 ClusterStreamChunk last = operationQueue.commitLast();
306 LOGGER.error("Trying to commit operation after MainProgram is closed! Operation is " + last);
312 * Called by DB client write threads
314 void schedule(ClusterUpdateOperation operation) throws IllegalAcornStateException {
317 LOGGER.error("Trying to schedule operation after MainProgram is closed! Operation is " + operation);
320 clusters.streamLRU.acquireMutex();
324 operationQueue.scheduleUpdate(operation);
327 } catch (IllegalAcornStateException e) {
329 } catch (Throwable t) {
330 throw new IllegalAcornStateException(t);
332 clusters.streamLRU.releaseMutex();
338 public void close() {
342 // This will wake up the sleeping beauty
343 operationQueue.scheduleTask(() -> {});
346 deathBarrier.acquire();
347 } catch (InterruptedException e) {
350 for (ExecutorService executor : clusterUpdateThreads)
353 for (int i = 0; i < clusterUpdateThreads.length; i++) {
355 ExecutorService executor = clusterUpdateThreads[i];
356 executor.awaitTermination(500, TimeUnit.MILLISECONDS);
357 clusterUpdateThreads[i] = null;
358 } catch (InterruptedException e) {
364 void assertMainProgramThread() {
365 assert(Thread.currentThread().equals(mainProgramThread));
368 void assertNoMainProgramThread() {
369 assert(!Thread.currentThread().equals(mainProgramThread));