774b605541371766333fa19957e57171bc40fdab
[simantics/platform.git] / bundles / org.simantics.acorn / src / org / simantics / acorn / GraphClientImpl2.java
1 /*******************************************************************************
2  * Copyright (c) 2007, 2010 Association for Decentralized Information Management
3  * in Industry THTH ry.
4  * All rights reserved. This program and the accompanying materials
5  * are made available under the terms of the Eclipse Public License v1.0
6  * which accompanies this distribution, and is available at
7  * http://www.eclipse.org/legal/epl-v10.html
8  *
9  * Contributors:
10  *     VTT Technical Research Centre of Finland - initial API and implementation
11  *******************************************************************************/
12 package org.simantics.acorn;
13
14 import java.io.IOException;
15 import java.nio.file.Path;
16 import java.util.ArrayList;
17 import java.util.LinkedList;
18 import java.util.concurrent.ExecutorService;
19 import java.util.concurrent.Executors;
20 import java.util.concurrent.Semaphore;
21 import java.util.concurrent.ThreadFactory;
22 import java.util.concurrent.TimeUnit;
23
24 import org.simantics.acorn.internal.ClusterChange;
25 import org.simantics.acorn.internal.ClusterUpdateProcessorBase;
26 import org.simantics.acorn.internal.UndoClusterUpdateProcessor;
27 import org.simantics.acorn.lru.ClusterInfo;
28 import org.simantics.acorn.lru.ClusterStreamChunk;
29 import org.simantics.acorn.lru.ClusterUpdateOperation;
30 import org.simantics.acorn.lru.ClusterChangeSet.Entry;
31 import org.simantics.db.ClusterCreator;
32 import org.simantics.db.Database;
33 import org.simantics.db.ServiceLocator;
34 import org.simantics.db.common.utils.Logger;
35 import org.simantics.db.exception.DatabaseException;
36 import org.simantics.db.exception.SDBException;
37 import org.simantics.db.server.ProCoreException;
38 import org.simantics.db.service.ClusterSetsSupport;
39 import org.simantics.db.service.ClusterUID;
40 import org.simantics.utils.datastructures.Pair;
41 import org.simantics.utils.logging.TimeLogger;
42
43 import gnu.trove.map.hash.TLongObjectHashMap;
44
45 public class GraphClientImpl2 implements Database.Session {
46         
47         public static final boolean DEBUG = false;
48
49         public final ClusterManager clusters;
50         
51         private TransactionManager transactionManager = new TransactionManager();
52         private ExecutorService executor = Executors.newSingleThreadExecutor(new ClientThreadFactory("Core Main Program", false));
53         private ExecutorService saver = Executors.newSingleThreadExecutor(new ClientThreadFactory("Core Snapshot Saver", true));
54
55         private static GraphClientImpl2 INSTANCE;
56         private Path dbFolder;
57         private final Database database;
58         private ServiceLocator locator;
59         private MainProgram mainProgram;
60
61         static class ClientThreadFactory implements ThreadFactory {
62                 
63                 final String name;
64                 final boolean daemon;
65                 
66                 public ClientThreadFactory(String name, boolean daemon) {
67                         this.name = name;
68                         this.daemon = daemon;
69                 }
70                 
71                 @Override
72                 public Thread newThread(Runnable r) {
73                         Thread thread = new Thread(r, name);
74                         thread.setDaemon(daemon);
75                         return thread;
76                 }
77         }
78
79         public GraphClientImpl2(Database database, Path dbFolder, ServiceLocator locator) throws IOException {
80             this.database = database;
81             this.dbFolder = dbFolder;
82             this.locator = locator;
83             this.clusters = new ClusterManager(dbFolder);
84             load();
85             ClusterSetsSupport cssi = locator.getService(ClusterSetsSupport.class); 
86             cssi.updateReadAndWriteDirectories(clusters.lastSessionDirectory, clusters.workingDirectory);
87             mainProgram = new MainProgram(this, clusters);
88             executor.execute(mainProgram);
89             INSTANCE = this;
90         }
91
92         public Path getDbFolder() {
93             return dbFolder;
94         }
95
96         public void tryMakeSnapshot() throws IOException {
97                 
98             if (isClosing)
99                 return;
100             
101                 saver.execute(new Runnable() {
102
103                         @Override
104                         public void run() {
105                                 Transaction tr = null;
106                                 try {
107                                         // First take a write transaction
108                                         tr = askWriteTransaction(-1);
109                                         // Then make sure that MainProgram is idling
110                                         mainProgram.mutex.acquire();
111                                         try {
112                                                 synchronized(mainProgram) {
113                                                         if(mainProgram.operations.isEmpty()) {
114                                                                 makeSnapshot(false);
115                                                         } else {
116                                                                 // MainProgram is becoming busy again - delay snapshotting
117                                                                 return;
118                                                         }
119                                                 }
120                                         } finally {
121                                                 mainProgram.mutex.release();
122                                         }
123                                 } catch (IOException e) {
124                                         Logger.defaultLogError(e);
125                                 } catch (ProCoreException e) {
126                                         Logger.defaultLogError(e);
127                                 } catch (InterruptedException e) {
128                                         Logger.defaultLogError(e);
129                                 } finally {
130                                         try {
131                                                 if(tr != null)
132                                                         endTransaction(tr.getTransactionId());
133                                         } catch (ProCoreException e) {
134                                                 Logger.defaultLogError(e);
135                                         }
136                                 }
137                         }
138                         
139                 });
140         }
141         
142     public void makeSnapshot(boolean force) throws IOException {
143         if (safeToMakeSnapshot)
144             clusters.makeSnapshot(locator, force);
145     }
146         
147         public <T> T clone(ClusterUID uid, ClusterCreator creator) throws DatabaseException {
148                 return clusters.clone(uid, creator);
149         }
150
151 //      private void save() throws IOException {
152 //              clusters.save();
153 //      }
154         
155         public void load() throws IOException {
156                 clusters.load();
157         }
158         
159 //      public void modiFileEx(ClusterUID uid, int resourceKey, long offset, long size, byte[] bytes, long pos, ClusterSupport support) {
160 //              clusters.modiFileEx(uid, resourceKey, offset, size, bytes, pos, support);
161 //      }
162
163         @Override
164         public Database getDatabase() {
165                 return database;
166         }
167
168         private boolean closed = false;
169         private boolean isClosing = false;
170         // Add check to make sure if it safe to make snapshot (used with cancel which is not yet supported and may cause corrupted head.state writing)
171     private boolean safeToMakeSnapshot = true;
172         
173         @Override
174         public void close() throws ProCoreException {
175             System.err.println("Closing " + this + " and mainProgram " + mainProgram);
176                 if(!closed && !isClosing) {
177                     isClosing = true;
178                         try {
179                             makeSnapshot(true);
180                                 
181                                 mainProgram.close();
182                                 clusters.shutdown();
183                                 executor.shutdown();
184                                 saver.shutdown();
185                                 boolean executorTerminated = executor.awaitTermination(500, TimeUnit.MILLISECONDS);
186                                 boolean saverTerminated = saver.awaitTermination(500, TimeUnit.MILLISECONDS);
187                                 
188                                 System.err.println("executorTerminated=" + executorTerminated + ", saverTerminated=" + saverTerminated);
189                                 
190                                 INSTANCE = null;
191                                 mainProgram = null;
192                                 executor = null;
193                                 saver = null;
194                                 
195                         } catch (IOException | InterruptedException e) {
196                                 throw new ProCoreException(e);
197                         }
198                 }
199                 closed = true;
200                 //impl.close();
201         }
202
203         @Override
204         public void open() throws ProCoreException {
205                 throw new UnsupportedOperationException();
206         }
207
208         @Override
209         public boolean isClosed() throws ProCoreException {
210                 return closed;
211         }
212         
213         @Override
214         public void acceptCommit(long transactionId, long changeSetId, byte[] metadata) throws ProCoreException {
215                 
216                 clusters.state.headChangeSetId++;
217
218                 long committedChangeSetId = changeSetId + 1;
219                 
220                 clusters.commitChangeSet(committedChangeSetId, metadata);
221                 
222                 clusters.state.transactionId = transactionId;
223                 
224                 mainProgram.committed();
225                 
226                 TimeLogger.log("Accepted commit");
227                 
228         }
229
230         @Override
231         public long cancelCommit(long transactionId, long changeSetId,
232                         byte[] metadata, OnChangeSetUpdate onChangeSetUpdate)
233                         throws ProCoreException {
234             safeToMakeSnapshot = false;
235             throw new UnsupportedOperationException("org.simantics.acorn.GraphClientImpl2.cancelCommit() is not supported operation! Closing down to prevent further havoc");
236 //          System.err.println("GraphClientImpl2.cancelCommit() called!! this is experimental and might cause havoc!");
237 //          try {
238 //            undo(new long[] {changeSetId}, onChangeSetUpdate);
239 //        } catch (SDBException e) {
240 //            e.printStackTrace();
241 //            throw new ProCoreException(e);
242 //        }
243 //          clusters.state.headChangeSetId++;
244 //          return clusters.state.headChangeSetId;
245         }
246
247         @Override
248         public Transaction askReadTransaction() throws ProCoreException {
249                 return transactionManager.askReadTransaction();
250         }
251
252         enum TransactionState {
253                 IDLE,WRITE,READ
254         }
255         
256         class TransactionRequest {
257                 public TransactionState state;
258                 public Semaphore semaphore;
259                 public TransactionRequest(TransactionState state, Semaphore semaphore) {
260                         this.state = state;
261                         this.semaphore = semaphore;
262                 }
263         }
264
265         class TransactionManager {
266
267                 private TransactionState currentTransactionState = TransactionState.IDLE;
268                 
269                 private int reads = 0;
270                 
271                 LinkedList<TransactionRequest> requests = new LinkedList<TransactionRequest>();
272                 
273                 TLongObjectHashMap<TransactionRequest> requestMap = new TLongObjectHashMap<TransactionRequest>();
274                 
275                 private synchronized Transaction makeTransaction(TransactionRequest req) {
276                         
277                         final int csId = clusters.state.headChangeSetId;
278                         final long trId = clusters.state.transactionId+1;
279                         requestMap.put(trId, req);
280                         return new Transaction() {
281                                 
282                                 @Override
283                                 public long getTransactionId() {
284                                         return trId;
285                                 }
286                                 
287                                 @Override
288                                 public long getHeadChangeSetId() {
289                                         return csId;
290                                 }
291                         };
292                 }
293                 
294                 /*
295                  * This method cannot be synchronized since it waits and must support multiple entries
296                  * by query thread(s) and internal transactions such as snapshot saver
297                  */
298                 public Transaction askReadTransaction() throws ProCoreException {
299                 
300                         Semaphore semaphore = new Semaphore(0);
301                         
302                         TransactionRequest req = queue(TransactionState.READ, semaphore);
303                         
304                         try {
305                                 semaphore.acquire();
306                         } catch (InterruptedException e) {
307                                 throw new ProCoreException(e);
308                         }
309                         
310                         return makeTransaction(req);
311
312                 }
313
314                 private synchronized void dispatch() {
315                         TransactionRequest r = requests.removeFirst();
316                         if(r.state == TransactionState.READ) reads++;
317                         r.semaphore.release();
318                 }
319                 
320                 private synchronized void processRequests() {
321                         
322                         while(true) {
323
324                                 if(requests.isEmpty()) return;
325                                 TransactionRequest req = requests.peek();
326
327                                 if(currentTransactionState == TransactionState.IDLE) {
328                                 
329                                         // Accept anything while IDLE
330                                         currentTransactionState = req.state;
331                                         dispatch();
332                                         
333                                 } else if (currentTransactionState == TransactionState.READ) {
334                                         
335                                         if(req.state == currentTransactionState) {
336
337                                                 // Allow other reads
338                                                 dispatch();
339
340                                         } else {
341                                                 
342                                                 // Wait
343                                                 return;
344                                                 
345                                         }
346                                         
347                                 }  else if (currentTransactionState == TransactionState.WRITE) {
348
349                                         // Wait
350                                         return;
351                                         
352                                 }
353                                 
354                         }
355                         
356                 }
357                 
358                 private synchronized TransactionRequest queue(TransactionState state, Semaphore semaphore) {
359                         TransactionRequest req = new TransactionRequest(state, semaphore); 
360                         requests.addLast(req);
361                         processRequests();
362                         return req;
363                 }
364                 
365                 /*
366                  * This method cannot be synchronized since it waits and must support multiple entries
367                  * by query thread(s) and internal transactions such as snapshot saver
368                  */
369                 public Transaction askWriteTransaction()
370                                 throws ProCoreException {
371                         
372                         Semaphore semaphore = new Semaphore(0);
373                         
374                         TransactionRequest req = queue(TransactionState.WRITE, semaphore);
375                         
376                         try {
377                                 semaphore.acquire();
378                         } catch (InterruptedException e) {
379                                 throw new ProCoreException(e);
380                         }
381                         
382                         mainProgram.startTransaction(clusters.state.headChangeSetId+1);
383                         
384                         return makeTransaction(req);
385                         
386                 }
387                 
388                 public synchronized long endTransaction(long transactionId) throws ProCoreException {
389                         
390                         TransactionRequest req = requestMap.remove(transactionId);
391                         if(req.state == TransactionState.WRITE) {
392                                 currentTransactionState = TransactionState.IDLE;
393                                 processRequests();
394                         } else {
395                                 reads--;
396                                 if(reads == 0) {
397                                         currentTransactionState = TransactionState.IDLE;
398                                         processRequests();
399                                 }
400                         }
401                         return clusters.state.transactionId;
402                 }
403
404         }
405         
406         @Override
407         public Transaction askWriteTransaction(final long transactionId)
408                         throws ProCoreException {
409                 return transactionManager.askWriteTransaction();
410         }
411
412         @Override
413         public long endTransaction(long transactionId) throws ProCoreException {
414                 return transactionManager.endTransaction(transactionId);
415         }
416
417         @Override
418         public String execute(String command) throws ProCoreException {
419                 // This is called only by WriteGraphImpl.commitAccessorChanges
420                 // We can ignore this in Acorn
421                 return "";
422         }
423
424         @Override
425         public byte[] getChangeSetMetadata(long changeSetId)
426                         throws ProCoreException {
427                 return clusters.getMetadata(changeSetId);
428         }
429
430         @Override
431         public ChangeSetData getChangeSetData(long minChangeSetId,
432                         long maxChangeSetId, OnChangeSetUpdate onChangeSetupate)
433                         throws ProCoreException {
434                 
435                 new Exception("GetChangeSetDataFunction " + minChangeSetId + " " + maxChangeSetId).printStackTrace();;
436                 return null;
437                 
438         }
439
440         @Override
441         public ChangeSetIds getChangeSetIds() throws ProCoreException {
442                 throw new UnsupportedOperationException();
443         }
444
445         @Override
446         public Cluster getCluster(byte[] clusterId) throws ProCoreException {
447                 throw new UnsupportedOperationException();
448         }
449
450         @Override
451         public ClusterChanges getClusterChanges(long changeSetId, byte[] clusterId)
452                         throws ProCoreException {
453                 throw new UnsupportedOperationException();
454         }
455
456         @Override
457         public ClusterIds getClusterIds() throws ProCoreException {
458                 return clusters.getClusterIds();
459         }
460
461         @Override
462         public Information getInformation() throws ProCoreException {
463                 return new Information() {
464
465                         @Override
466                         public String getServerId() {
467                                 return "server";
468                         }
469
470                         @Override
471                         public String getProtocolId() {
472                                 return "";
473                         }
474
475                         @Override
476                         public String getDatabaseId() {
477                                 return "database";
478                         }
479
480                         @Override
481                         public long getFirstChangeSetId() {
482                                 return 0;
483                         }
484                         
485                 };
486         }
487
488         @Override
489         public Refresh getRefresh(long changeSetId) throws ProCoreException {
490                 
491                 final ClusterIds ids = getClusterIds();
492                 
493                 return new Refresh() {
494
495                         @Override
496                         public long getHeadChangeSetId() {
497                                 return clusters.state.headChangeSetId;
498                         }
499
500                         @Override
501                         public long[] getFirst() {
502                                 return ids.getFirst();
503                         }
504
505                         @Override
506                         public long[] getSecond() {
507                                 return ids.getSecond();
508                         }
509                         
510                 };
511                 
512         }
513
514         public byte[] getResourceFile(final byte[] clusterUID, final int resourceIndex) throws ProCoreException {
515                 return clusters.getResourceFile(clusterUID, resourceIndex);
516         }
517
518         @Override
519         public ResourceSegment getResourceSegment(final byte[] clusterUID,
520                         final int resourceIndex, final long segmentOffset, short segmentSize) throws ProCoreException {
521                 
522                 return clusters.getResourceSegment(clusterUID, resourceIndex, segmentOffset, segmentSize);
523
524         }
525
526         @Override
527         public long reserveIds(int count) throws ProCoreException {
528                 return clusters.state.reservedIds++;
529         }
530
531         @Override
532         public void updateCluster(byte[] operations) throws ProCoreException {
533
534                 ClusterUpdateOperation operation = new ClusterUpdateOperation(clusters, operations);
535                 ClusterInfo info = clusters.clusterLRU.getOrCreate(operation.uid, true);
536                 if(info == null) throw new IllegalStateException();
537                 info.acquireMutex();
538                 try {
539                         info.scheduleUpdate();
540                         mainProgram.schedule(operation);
541                 } catch (Throwable t) {
542                         throw new IllegalStateException(t);
543                 } finally {
544                         info.releaseMutex();
545                 }
546
547         }
548
549         private UndoClusterUpdateProcessor getUndoCSS(String ccsId) throws DatabaseException {
550
551                 String[] ss = ccsId.split("\\.");
552                 String chunkKey = ss[0];
553                 int chunkOffset = Integer.parseInt(ss[1]);
554                 ClusterStreamChunk chunk = clusters.streamLRU.getWithoutMutex(chunkKey);
555                 if(chunk == null) throw new IllegalStateException("Cluster Stream Chunk " + chunkKey + " was not found.");
556                 chunk.acquireMutex();
557                 try {
558                         return chunk.getUndoProcessor(clusters, chunkOffset, ccsId);
559                 } catch (Throwable t) {
560                         throw new IllegalStateException(t);
561                 } finally {
562                         chunk.releaseMutex();
563                 }
564                 
565         }
566         
567         private void performUndo(String ccsId, ArrayList<Pair<ClusterUID, byte[]>> clusterChanges, UndoClusterSupport support) throws ProCoreException, DatabaseException {
568
569                 UndoClusterUpdateProcessor proc = getUndoCSS(ccsId);
570
571                 int clusterKey = clusters.getClusterKeyByClusterUIDOrMakeWithoutMutex(proc.getClusterUID());
572
573                 clusters.clusterLRU.acquireMutex();
574                 try {
575
576                         ClusterChange cs = new ClusterChange(clusterChanges, proc.getClusterUID());
577                         for(int i=0;i<proc.entries.size();i++) {
578                                 
579                                 Entry e = proc.entries.get(proc.entries.size() - 1 - i);
580                                 e.process(clusters, cs, clusterKey);
581                                 
582                         }
583                         
584                         cs.flush();
585
586                 } finally {
587                         clusters.clusterLRU.releaseMutex();
588                 }
589                 
590         }
591         
592         @Override
593         public boolean undo(long[] changeSetIds, OnChangeSetUpdate onChangeSetUpdate) throws SDBException {
594                 
595                 final ArrayList<Pair<ClusterUID, byte[]>> clusterChanges = new ArrayList<Pair<ClusterUID, byte[]>>();
596                 
597                 UndoClusterSupport support = new UndoClusterSupport(clusters);
598                 
599                 final int changeSetId = clusters.state.headChangeSetId;
600                 
601                 if(ClusterUpdateProcessorBase.DEBUG)
602                         System.err.println(" === BEGIN UNDO ===");
603                 
604                 for(int i=0;i<changeSetIds.length;i++) {
605                         final long id = changeSetIds[changeSetIds.length-1-i];
606                         ArrayList<String> ccss = clusters.getChanges(id);
607                         for(int j=0;j<ccss.size();j++) {
608                                 try {
609                                         if(ClusterUpdateProcessorBase.DEBUG)
610                                                 System.err.println("performUndo " + ccss.get(ccss.size()-j-1));
611                                         performUndo(ccss.get(ccss.size()-j-1), clusterChanges, support);
612                                 } catch (DatabaseException e) {
613                                         e.printStackTrace();
614                                 }
615                         }
616                 }
617
618                 if(ClusterUpdateProcessorBase.DEBUG)
619                         System.err.println(" === END UNDO ===");
620
621                 for(int i=0;i<clusterChanges.size();i++) {
622                         
623                         final int changeSetIndex = i;
624                         
625                         final Pair<ClusterUID, byte[]> pair = clusterChanges.get(i);
626                         
627                         final ClusterUID cuid = pair.first;
628                         final byte[] data = pair.second;
629
630                         onChangeSetUpdate.onChangeSetUpdate(new ChangeSetUpdate() {
631
632                                 @Override
633                                 public long getChangeSetId() {
634                                         return changeSetId;
635                                 }
636
637                                 @Override
638                                 public int getChangeSetIndex() {
639                                         return 0;
640                                 }
641
642                                 @Override
643                                 public int getNumberOfClusterChangeSets() {
644                                         return clusterChanges.size();
645                                 }
646
647                                 @Override
648                                 public int getIndexOfClusterChangeSet() {
649                                         return changeSetIndex;
650                                 }
651
652                                 @Override
653                                 public byte[] getClusterId() {
654                                         return cuid.asBytes();
655                                 }
656
657                                 @Override
658                                 public boolean getNewCluster() {
659                                         return false;
660                                 }
661
662                                 @Override
663                                 public byte[] getData() {
664                                         return data;
665                                 }
666
667                         });
668
669                 }
670
671                 
672                 return false;
673                 
674         }
675         
676         public static GraphClientImpl2 getInstance() {
677             return INSTANCE;
678         }
679         
680         public ServiceLocator getServiceLocator() {
681             return locator;
682         }
683
684     @Override
685     public boolean refreshEnabled() {
686         return false;
687     }
688
689         
690         
691         
692         
693         
694         
695         
696         
697         
698         
699         ////////////////////////
700         
701         
702         
703         
704         
705         
706         
707         
708         
709         
710         
711         
712 }
713